Whamcloud - gitweb
Branch HEAD
[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 [sun.com URL with a
20  * copy of GPLv2].
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                 loghandle->lgh_last_idx++;
328                 rc = llog_lvfs_pad(obd, file, left, loghandle->lgh_last_idx);
329                 if (rc)
330                         RETURN(rc);
331                 /* if it's the last idx in log file, then return -ENOSPC */
332                 if (loghandle->lgh_last_idx == LLOG_BITMAP_SIZE(llh) - 1)
333                         RETURN(-ENOSPC);
334         }
335
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         if (ext2_set_bit(index, llh->llh_bitmap)) {
347                 CERROR("argh, index %u already set in log bitmap?\n", index);
348                 LBUG(); /* should never happen */
349         }
350         llh->llh_count++;
351         llh->llh_tail.lrt_index = index;
352
353         rc = llog_lvfs_write_blob(obd, file, &llh->llh_hdr, NULL, 0);
354         if (rc)
355                 RETURN(rc);
356
357         rc = llog_lvfs_write_blob(obd, file, rec, buf, file->f_pos);
358         if (rc)
359                 RETURN(rc);
360
361         CDEBUG(D_RPCTRACE, "added record "LPX64": idx: %u, %u bytes\n",
362                loghandle->lgh_id.lgl_oid, index, rec->lrh_len);
363         if (rc == 0 && reccookie) {
364                 reccookie->lgc_lgl = loghandle->lgh_id;
365                 reccookie->lgc_index = index;
366                 if ((rec->lrh_type == MDS_UNLINK_REC) || 
367                                 (rec->lrh_type == MDS_SETATTR_REC))
368                         reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
369                 else if (rec->lrh_type == OST_SZ_REC)
370                         reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
371                 else if (rec->lrh_type == OST_RAID1_REC)
372                         reccookie->lgc_subsys = LLOG_RD1_ORIG_CTXT;
373                 else
374                         reccookie->lgc_subsys = -1;
375                 rc = 1;
376         }
377         if (rc == 0 && rec->lrh_type == LLOG_GEN_REC)
378                 rc = 1;
379
380         RETURN(rc);
381 }
382
383 /* We can skip reading at least as many log blocks as the number of
384 * minimum sized log records we are skipping.  If it turns out
385 * that we are not far enough along the log (because the
386 * actual records are larger than minimum size) we just skip
387 * some more records. */
388
389 static void llog_skip_over(__u64 *off, int curr, int goal)
390 {
391         if (goal <= curr)
392                 return;
393         *off = (*off + (goal-curr-1) * LLOG_MIN_REC_SIZE) &
394                 ~(LLOG_CHUNK_SIZE - 1);
395 }
396
397
398 /* sets:
399  *  - cur_offset to the furthest point read in the log file
400  *  - cur_idx to the log index preceeding cur_offset
401  * returns -EIO/-EINVAL on error
402  */
403 static int llog_lvfs_next_block(struct llog_handle *loghandle, int *cur_idx,
404                                 int next_idx, __u64 *cur_offset, void *buf,
405                                 int len)
406 {
407         int rc;
408         ENTRY;
409
410         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
411                 RETURN(-EINVAL);
412
413         CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off "LPU64")\n",
414                next_idx, *cur_idx, *cur_offset);
415
416         while (*cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
417                 struct llog_rec_hdr *rec;
418                 struct llog_rec_tail *tail;
419                 loff_t ppos;
420
421                 llog_skip_over(cur_offset, *cur_idx, next_idx);
422
423                 ppos = *cur_offset;
424                 rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
425                                         loghandle->lgh_file, buf, len,
426                                         &ppos);
427                 if (rc) {
428                         CERROR("Cant read llog block at log id "LPU64
429                                "/%u offset "LPU64"\n",
430                                loghandle->lgh_id.lgl_oid,
431                                loghandle->lgh_id.lgl_ogen,
432                                *cur_offset);
433                         RETURN(rc);
434                 }
435
436                 /* put number of bytes read into rc to make code simpler */
437                 rc = ppos - *cur_offset;
438                 *cur_offset = ppos;
439
440                 if (rc < len) {
441                         /* signal the end of the valid buffer to llog_process */
442                         memset(buf + rc, 0, len - rc);
443                 }
444
445                 if (rc == 0) /* end of file, nothing to do */
446                         RETURN(0);
447
448                 if (rc < sizeof(*tail)) {
449                         CERROR("Invalid llog block at log id "LPU64"/%u offset "
450                                LPU64"\n", loghandle->lgh_id.lgl_oid,
451                                loghandle->lgh_id.lgl_ogen, *cur_offset);
452                         RETURN(-EINVAL);
453                 }
454
455                 rec = buf;
456                 tail = (struct llog_rec_tail *)((char *)buf + rc -
457                                                 sizeof(struct llog_rec_tail));
458
459                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec)) {
460                         lustre_swab_llog_rec(rec, tail);
461                 }
462
463                 *cur_idx = tail->lrt_index;
464
465                 /* this shouldn't happen */
466                 if (tail->lrt_index == 0) {
467                         CERROR("Invalid llog tail at log id "LPU64"/%u offset "
468                                LPU64"\n", loghandle->lgh_id.lgl_oid,
469                                loghandle->lgh_id.lgl_ogen, *cur_offset);
470                         RETURN(-EINVAL);
471                 }
472                 if (tail->lrt_index < next_idx)
473                         continue;
474
475                 /* sanity check that the start of the new buffer is no farther
476                  * than the record that we wanted.  This shouldn't happen. */
477                 if (rec->lrh_index > next_idx) {
478                         CERROR("missed desired record? %u > %u\n",
479                                rec->lrh_index, next_idx);
480                         RETURN(-ENOENT);
481                 }
482                 RETURN(0);
483         }
484         RETURN(-EIO);
485 }
486
487 static int llog_lvfs_prev_block(struct llog_handle *loghandle,
488                                 int prev_idx, void *buf, int len)
489 {
490         __u64 cur_offset;
491         int rc;
492         ENTRY;
493
494         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
495                 RETURN(-EINVAL);
496
497         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
498
499         cur_offset = LLOG_CHUNK_SIZE;
500         llog_skip_over(&cur_offset, 0, prev_idx);
501
502         while (cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
503                 struct llog_rec_hdr *rec;
504                 struct llog_rec_tail *tail;
505                 loff_t ppos;
506
507                 ppos = cur_offset;
508
509                 rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
510                                         loghandle->lgh_file, buf, len,
511                                         &ppos);
512                 if (rc) {
513                         CERROR("Cant read llog block at log id "LPU64
514                                "/%u offset "LPU64"\n",
515                                loghandle->lgh_id.lgl_oid,
516                                loghandle->lgh_id.lgl_ogen,
517                                cur_offset);
518                         RETURN(rc);
519                 }
520
521                 /* put number of bytes read into rc to make code simpler */
522                 rc = ppos - cur_offset;
523                 cur_offset = ppos;
524
525                 if (rc == 0) /* end of file, nothing to do */
526                         RETURN(0);
527
528                 if (rc < sizeof(*tail)) {
529                         CERROR("Invalid llog block at log id "LPU64"/%u offset "
530                                LPU64"\n", loghandle->lgh_id.lgl_oid,
531                                loghandle->lgh_id.lgl_ogen, cur_offset);
532                         RETURN(-EINVAL);
533                 }
534
535                 tail = buf + rc - sizeof(struct llog_rec_tail);
536
537                 /* this shouldn't happen */
538                 if (tail->lrt_index == 0) {
539                         CERROR("Invalid llog tail at log id "LPU64"/%u offset "
540                                LPU64"\n", loghandle->lgh_id.lgl_oid,
541                                loghandle->lgh_id.lgl_ogen, cur_offset);
542                         RETURN(-EINVAL);
543                 }
544                 if (le32_to_cpu(tail->lrt_index) < prev_idx)
545                         continue;
546
547                 /* sanity check that the start of the new buffer is no farther
548                  * than the record that we wanted.  This shouldn't happen. */
549                 rec = buf;
550                 if (le32_to_cpu(rec->lrh_index) > prev_idx) {
551                         CERROR("missed desired record? %u > %u\n",
552                                le32_to_cpu(rec->lrh_index), prev_idx);
553                         RETURN(-ENOENT);
554                 }
555                 RETURN(0);
556         }
557         RETURN(-EIO);
558 }
559
560 static struct file *llog_filp_open(char *dir, char *name, int flags, int mode)
561 {
562         char *logname;
563         struct file *filp;
564         int len;
565
566         OBD_ALLOC(logname, PATH_MAX);
567         if (logname == NULL)
568                 return ERR_PTR(-ENOMEM);
569
570         len = snprintf(logname, PATH_MAX, "%s/%s", dir, name);
571         if (len >= PATH_MAX - 1) {
572                 filp = ERR_PTR(-ENAMETOOLONG);
573         } else {
574                 filp = l_filp_open(logname, flags, mode);
575                 if (IS_ERR(filp))
576                         CERROR("logfile creation %s: %ld\n", logname,
577                                PTR_ERR(filp));
578         }
579         OBD_FREE(logname, PATH_MAX);
580         return filp;
581 }
582
583 /* This is a callback from the llog_* functions.
584  * Assumes caller has already pushed us into the kernel context. */
585 static int llog_lvfs_create(struct llog_ctxt *ctxt, struct llog_handle **res,
586                             struct llog_logid *logid, char *name)
587 {
588         struct llog_handle *handle;
589         struct obd_device *obd;
590         struct l_dentry *dchild = NULL;
591         struct obdo *oa = NULL;
592         int rc = 0, cleanup_phase = 1;
593         int open_flags = O_RDWR | O_CREAT | O_LARGEFILE;
594         ENTRY;
595
596         handle = llog_alloc_handle();
597         if (handle == NULL)
598                 RETURN(-ENOMEM);
599         *res = handle;
600
601         LASSERT(ctxt);
602         LASSERT(ctxt->loc_exp);
603         obd = ctxt->loc_exp->exp_obd;
604
605         if (logid != NULL) {
606                 dchild = obd_lvfs_fid2dentry(ctxt->loc_exp, logid->lgl_oid,
607                                              logid->lgl_ogen, logid->lgl_ogr);
608
609                 if (IS_ERR(dchild)) {
610                         rc = PTR_ERR(dchild);
611                         CERROR("error looking up logfile "LPX64":0x%x: rc %d\n",
612                                logid->lgl_oid, logid->lgl_ogen, rc);
613                         GOTO(cleanup, rc);
614                 }
615
616                 cleanup_phase = 2;
617                 if (dchild->d_inode == NULL) {
618                         rc = -ENOENT;
619                         CERROR("nonexistent log file "LPX64":"LPX64": rc %d\n",
620                                logid->lgl_oid, logid->lgl_ogr, rc);
621                         GOTO(cleanup, rc);
622                 }
623
624                 handle->lgh_file = l_dentry_open(&obd->obd_lvfs_ctxt, dchild,
625                                                     O_RDWR | O_LARGEFILE);
626                 if (IS_ERR(handle->lgh_file)) {
627                         rc = PTR_ERR(handle->lgh_file);
628                         CERROR("error opening logfile "LPX64"0x%x: rc %d\n",
629                                logid->lgl_oid, logid->lgl_ogen, rc);
630                         GOTO(cleanup, rc);
631                 }
632
633                 /* assign the value of lgh_id for handle directly */
634                 handle->lgh_id = *logid;
635
636         } else if (name) {
637                 /* COMPAT_146 */
638                 if (strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME) == 0) {
639                         handle->lgh_file = llog_filp_open(MDT_LOGS_DIR, name, 
640                                                           open_flags, 0644);
641                 } else {
642                         /* end COMPAT_146 */
643                         handle->lgh_file = llog_filp_open(MOUNT_CONFIGS_DIR,
644                                                           name, open_flags, 
645                                                           0644);
646                 }
647                 if (IS_ERR(handle->lgh_file))
648                         GOTO(cleanup, rc = PTR_ERR(handle->lgh_file));
649
650                 handle->lgh_id.lgl_ogr = 1;
651                 handle->lgh_id.lgl_oid =
652                         handle->lgh_file->f_dentry->d_inode->i_ino;
653                 handle->lgh_id.lgl_ogen =
654                         handle->lgh_file->f_dentry->d_inode->i_generation;
655         } else {
656                 OBDO_ALLOC(oa);
657                 if (oa == NULL)
658                         GOTO(cleanup, rc = -ENOMEM);
659
660                 oa->o_gr = FILTER_GROUP_LLOG;
661                 oa->o_valid = OBD_MD_FLGENER | OBD_MD_FLGROUP;
662
663                 rc = obd_create(ctxt->loc_exp, oa, NULL, NULL);
664                 if (rc)
665                         GOTO(cleanup, rc);
666
667                 dchild = obd_lvfs_fid2dentry(ctxt->loc_exp, oa->o_id,
668                                              oa->o_generation, oa->o_gr);
669
670                 if (IS_ERR(dchild))
671                         GOTO(cleanup, rc = PTR_ERR(dchild));
672                 cleanup_phase = 2;
673                 handle->lgh_file = l_dentry_open(&obd->obd_lvfs_ctxt, dchild,
674                                                  open_flags);
675                 if (IS_ERR(handle->lgh_file))
676                         GOTO(cleanup, rc = PTR_ERR(handle->lgh_file));
677
678                 handle->lgh_id.lgl_ogr = oa->o_gr;
679                 handle->lgh_id.lgl_oid = oa->o_id;
680                 handle->lgh_id.lgl_ogen = oa->o_generation;
681         }
682
683         handle->lgh_ctxt = ctxt;
684  finish:
685         if (oa)
686                 OBDO_FREE(oa);
687         RETURN(rc);
688 cleanup:
689         switch (cleanup_phase) {
690         case 2:
691                 l_dput(dchild);
692         case 1:
693                 llog_free_handle(handle);
694         }
695         goto finish;
696 }
697
698 static int llog_lvfs_close(struct llog_handle *handle)
699 {
700         int rc;
701         ENTRY;
702
703         rc = filp_close(handle->lgh_file, 0);
704         if (rc)
705                 CERROR("error closing log: rc %d\n", rc);
706         RETURN(rc);
707 }
708
709 static int llog_lvfs_destroy(struct llog_handle *handle)
710 {
711         struct dentry *fdentry;
712         struct obdo *oa;
713         struct obd_device *obd = handle->lgh_ctxt->loc_exp->exp_obd;
714         char *dir;
715         int rc;
716         ENTRY;
717
718         /* COMPAT_146 */
719         if (strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME) == 0)
720                 dir = MDT_LOGS_DIR;
721         else
722                 /* end COMPAT_146 */
723                 dir = MOUNT_CONFIGS_DIR;
724
725         fdentry = handle->lgh_file->f_dentry;
726         if (strcmp(fdentry->d_parent->d_name.name, dir) == 0) {
727                 struct inode *inode = fdentry->d_parent->d_inode;
728                 struct lvfs_run_ctxt saved;
729                 struct vfsmount *mnt = mntget(handle->lgh_file->f_vfsmnt);
730
731                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
732                 dget(fdentry);
733                 rc = llog_lvfs_close(handle);
734
735                 if (rc == 0) {
736                         LOCK_INODE_MUTEX(inode);
737                         rc = ll_vfs_unlink(inode, fdentry, mnt);
738                         UNLOCK_INODE_MUTEX(inode);
739                 }
740                 mntput(mnt);
741
742                 dput(fdentry);
743                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
744                 RETURN(rc);
745         }
746
747         OBDO_ALLOC(oa);
748         if (oa == NULL)
749                 RETURN(-ENOMEM);
750
751         oa->o_id = handle->lgh_id.lgl_oid;
752         oa->o_gr = handle->lgh_id.lgl_ogr;
753         oa->o_generation = handle->lgh_id.lgl_ogen;
754         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLGENER;
755
756         rc = llog_lvfs_close(handle);
757         if (rc)
758                 GOTO(out, rc);
759
760         rc = obd_destroy(handle->lgh_ctxt->loc_exp, oa, NULL, NULL, NULL);
761  out:
762         OBDO_FREE(oa);
763         RETURN(rc);
764 }
765
766 /* reads the catalog list */
767 int llog_get_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
768                       char *name, int count, struct llog_catid *idarray)
769 {
770         struct lvfs_run_ctxt saved;
771         struct l_file *file;
772         int rc;
773         int size = sizeof(*idarray) * count;
774         loff_t off = 0;
775         ENTRY;
776
777         if (!count) 
778                 RETURN(0);
779
780         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
781         file = filp_open(name, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
782         if (!file || IS_ERR(file)) {
783                 rc = PTR_ERR(file);
784                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
785                        name, rc);
786                 GOTO(out, rc);
787         }
788         
789         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
790                 CERROR("%s is not a regular file!: mode = %o\n", name,
791                        file->f_dentry->d_inode->i_mode);
792                 GOTO(out, rc = -ENOENT);
793         }
794
795         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
796                (int)i_size_read(file->f_dentry->d_inode), size);
797
798         rc = fsfilt_read_record(disk_obd, file, idarray, size, &off);
799         if (rc) {
800                 CERROR("OBD filter: error reading %s: rc %d\n", name, rc);
801                 GOTO(out, rc);
802         }
803
804         EXIT;
805  out:
806         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
807         if (file && !IS_ERR(file))
808                 rc = filp_close(file, 0);
809         return rc;
810 }
811 EXPORT_SYMBOL(llog_get_cat_list);
812
813 /* writes the cat list */
814 int llog_put_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
815                       char *name, int count, struct llog_catid *idarray)
816 {
817         struct lvfs_run_ctxt saved;
818         struct l_file *file;
819         int rc;
820         int size = sizeof(*idarray) * count;
821         loff_t off = 0;
822
823         if (!count) 
824                 return (0);
825
826         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
827         file = filp_open(name, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
828         if (!file || IS_ERR(file)) {
829                 rc = PTR_ERR(file);
830                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
831                        name, rc);
832                 GOTO(out, rc);
833         }
834
835         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
836                 CERROR("%s is not a regular file!: mode = %o\n", name,
837                        file->f_dentry->d_inode->i_mode);
838                 GOTO(out, rc = -ENOENT);
839         }
840
841         rc = fsfilt_write_record(disk_obd, file, idarray, size, &off, 1);
842         if (rc) {
843                 CDEBUG(D_INODE,"OBD filter: error reading %s: rc %d\n",
844                        name, rc);
845                 GOTO(out, rc);
846         }
847
848  out:
849         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
850         if (file && !IS_ERR(file))
851                 rc = filp_close(file, 0);
852         RETURN(rc);
853 }
854
855 struct llog_operations llog_lvfs_ops = {
856         lop_write_rec:   llog_lvfs_write_rec,
857         lop_next_block:  llog_lvfs_next_block,
858         lop_prev_block:  llog_lvfs_prev_block,
859         lop_read_header: llog_lvfs_read_header,
860         lop_create:      llog_lvfs_create,
861         lop_destroy:     llog_lvfs_destroy,
862         lop_close:       llog_lvfs_close,
863         //        lop_cancel: llog_lvfs_cancel,
864 };
865
866 EXPORT_SYMBOL(llog_lvfs_ops);
867
868 #else /* !__KERNEL__ */
869
870 static int llog_lvfs_read_header(struct llog_handle *handle)
871 {
872         LBUG();
873         return 0;
874 }
875
876 static int llog_lvfs_write_rec(struct llog_handle *loghandle,
877                                struct llog_rec_hdr *rec,
878                                struct llog_cookie *reccookie, int cookiecount,
879                                void *buf, int idx)
880 {
881         LBUG();
882         return 0;
883 }
884
885 static int llog_lvfs_next_block(struct llog_handle *loghandle, int *cur_idx,
886                                 int next_idx, __u64 *cur_offset, void *buf,
887                                 int len)
888 {
889         LBUG();
890         return 0;
891 }
892
893 static int llog_lvfs_prev_block(struct llog_handle *loghandle,
894                                 int prev_idx, void *buf, int len)
895 {
896         LBUG();
897         return 0;
898 }
899
900 static int llog_lvfs_create(struct llog_ctxt *ctxt, struct llog_handle **res,
901                             struct llog_logid *logid, char *name)
902 {
903         LBUG();
904         return 0;
905 }
906
907 static int llog_lvfs_close(struct llog_handle *handle)
908 {
909         LBUG();
910         return 0;
911 }
912
913 static int llog_lvfs_destroy(struct llog_handle *handle)
914 {
915         LBUG();
916         return 0;
917 }
918
919 int llog_get_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
920                       char *name, int count, struct llog_catid *idarray)
921 {
922         LBUG();
923         return 0;
924 }
925
926 int llog_put_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
927                       char *name, int count, struct llog_catid *idarray)
928 {
929         LBUG();
930         return 0;
931 }
932
933 struct llog_operations llog_lvfs_ops = {
934         lop_write_rec:   llog_lvfs_write_rec,
935         lop_next_block:  llog_lvfs_next_block,
936         lop_prev_block:  llog_lvfs_prev_block,
937         lop_read_header: llog_lvfs_read_header,
938         lop_create:      llog_lvfs_create,
939         lop_destroy:     llog_lvfs_destroy,
940         lop_close:       llog_lvfs_close,
941 //        lop_cancel:      llog_lvfs_cancel,
942 };
943 #endif