Whamcloud - gitweb
c7b65b31086f88fc6e807faca596091af5099ad0
[fs/lustre-release.git] / lustre / obdfilter / filter.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/obdfilter/filter.c
37  *
38  * Author: Peter Braam <braam@clusterfs.com>
39  * Author: Andreas Dilger <adilger@clusterfs.com>
40  */
41
42 /*
43  * Invariant: Get O/R i_mutex for lookup, if needed, before any journal ops
44  *            (which need to get journal_lock, may block if journal full).
45  *
46  * Invariant: Call filter_start_transno() before any journal ops to avoid the
47  *            same deadlock problem.  We can (and want) to get rid of the
48  *            transno sem in favour of the dir/inode i_mutex to avoid single
49  *            threaded operation on the OST.
50  */
51
52 #define DEBUG_SUBSYSTEM S_FILTER
53
54 #ifndef AUTOCONF_INCLUDED
55 #include <linux/config.h>
56 #endif
57 #include <linux/module.h>
58 #include <linux/fs.h>
59 #include <linux/dcache.h>
60 #include <linux/init.h>
61 #include <linux/version.h>
62 #include <linux/sched.h>
63 #include <linux/mount.h>
64 #include <linux/buffer_head.h>
65
66 #include <obd_cksum.h>
67 #include <obd_class.h>
68 #include <obd_lov.h>
69 #include <lustre_dlm.h>
70 #include <lustre_fsfilt.h>
71 #include <lprocfs_status.h>
72 #include <lustre_log.h>
73 #include <libcfs/list.h>
74 #include <lustre_disk.h>
75 #include <lustre_quota.h>
76 #include <linux/slab.h>
77 #include <lustre_param.h>
78 #include <lustre/ll_fiemap.h>
79
80 #include "filter_internal.h"
81
82 /* Group 0 is no longer a legal group, to catch uninitialized IDs */
83 #define FILTER_MIN_GROUPS FILTER_GROUP_MDS0
84 static struct lvfs_callback_ops filter_lvfs_ops;
85 cfs_mem_cache_t *ll_fmd_cachep;
86
87 static void filter_commit_cb(struct obd_device *obd, __u64 transno,
88                              void *cb_data, int error)
89 {
90         obd_transno_commit_cb(obd, transno, error);
91 }
92
93 /* Assumes caller has already pushed us into the kernel context. */
94 int filter_finish_transno(struct obd_export *exp, struct obd_trans_info *oti,
95                           int rc, int force_sync)
96 {
97         struct filter_obd *filter = &exp->exp_obd->u.filter;
98         struct filter_export_data *fed = &exp->exp_filter_data;
99         struct lsd_client_data *lcd = fed->fed_lcd;
100         __u64 last_rcvd;
101         loff_t off;
102         int err, log_pri = D_RPCTRACE;
103
104         /* Propagate error code. */
105         if (rc)
106                 RETURN(rc);
107
108         if (!exp->exp_obd->obd_replayable || oti == NULL)
109                 RETURN(rc);
110
111         /* we don't allocate new transnos for replayed requests */
112         if (oti->oti_transno == 0) {
113                 spin_lock(&filter->fo_translock);
114                 last_rcvd = le64_to_cpu(filter->fo_fsd->lsd_last_transno) + 1;
115                 filter->fo_fsd->lsd_last_transno = cpu_to_le64(last_rcvd);
116                 spin_unlock(&filter->fo_translock);
117                 oti->oti_transno = last_rcvd;
118         } else {
119                 spin_lock(&filter->fo_translock);
120                 last_rcvd = oti->oti_transno;
121                 if (last_rcvd > le64_to_cpu(filter->fo_fsd->lsd_last_transno))
122                         filter->fo_fsd->lsd_last_transno =
123                                 cpu_to_le64(last_rcvd);
124                 spin_unlock(&filter->fo_translock);
125         }
126         lcd->lcd_last_transno = cpu_to_le64(last_rcvd);
127
128         /* could get xid from oti, if it's ever needed */
129         lcd->lcd_last_xid = 0;
130
131         off = fed->fed_lr_off;
132         if (off <= 0) {
133                 CERROR("%s: client idx %d is %lld\n", exp->exp_obd->obd_name,
134                        fed->fed_lr_idx, fed->fed_lr_off);
135                 err = -EINVAL;
136         } else {
137                 if (!force_sync)
138                         force_sync = fsfilt_add_journal_cb(exp->exp_obd,
139                                                            last_rcvd,
140                                                            oti->oti_handle,
141                                                            filter_commit_cb,
142                                                            NULL);
143
144                 err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp,
145                                           lcd, sizeof(*lcd), &off,
146                                           force_sync | exp->exp_need_sync);
147                 if (force_sync)
148                         filter_commit_cb(exp->exp_obd, last_rcvd, NULL, err);
149         }
150         if (err) {
151                 log_pri = D_ERROR;
152                 if (rc == 0)
153                         rc = err;
154         }
155
156         CDEBUG(log_pri, "wrote trans "LPU64" for client %s at #%d: err = %d\n",
157                last_rcvd, lcd->lcd_uuid, fed->fed_lr_idx, err);
158
159         RETURN(rc);
160 }
161
162 void f_dput(struct dentry *dentry)
163 {
164         /* Can't go inside filter_ddelete because it can block */
165         CDEBUG(D_INODE, "putting %s: %p, count = %d\n",
166                dentry->d_name.name, dentry, atomic_read(&dentry->d_count) - 1);
167         LASSERT(atomic_read(&dentry->d_count) > 0);
168
169         dput(dentry);
170 }
171
172 static void init_brw_stats(struct brw_stats *brw_stats)
173 {
174         int i;
175         for (i = 0; i < BRW_LAST; i++)
176                 spin_lock_init(&brw_stats->hist[i].oh_lock);
177 }
178
179 static int lprocfs_init_rw_stats(struct obd_device *obd,
180                                  struct lprocfs_stats **stats)
181 {
182         int num_stats;
183
184         num_stats = (sizeof(*obd->obd_type->typ_dt_ops) / sizeof(void *)) +
185                                                         LPROC_FILTER_LAST - 1;
186         *stats = lprocfs_alloc_stats(num_stats, 0);
187         if (*stats == NULL)
188                 return -ENOMEM;
189
190         lprocfs_init_ops_stats(LPROC_FILTER_LAST, *stats);
191         lprocfs_counter_init(*stats, LPROC_FILTER_READ_BYTES,
192                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
193         lprocfs_counter_init(*stats, LPROC_FILTER_WRITE_BYTES,
194                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
195
196         return(0);
197 }
198
199 /* brw_stats are 2128, ops are 3916, ldlm are 204, so 6248 bytes per client,
200    plus the procfs overhead :( */
201 static int filter_export_stats_init(struct obd_device *obd,
202                                     struct obd_export *exp,
203                                     void *client_nid)
204 {
205         struct filter_export_data *fed = &exp->exp_filter_data;
206         int rc, newnid = 0;
207         ENTRY;
208
209         init_brw_stats(&fed->fed_brw_stats);
210
211         if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
212                 /* Self-export gets no proc entry */
213                 RETURN(0);
214
215         rc = lprocfs_exp_setup(exp, client_nid, &newnid);
216         if (rc) {
217                 /* Mask error for already created
218                  * /proc entries */
219                 if (rc == -EALREADY)
220                         rc = 0;
221                 RETURN(rc);
222         }
223
224         if (newnid) {
225                 struct nid_stat *tmp = exp->exp_nid_stats;
226                 LASSERT(tmp != NULL);
227
228                 OBD_ALLOC(tmp->nid_brw_stats, sizeof(struct brw_stats));
229                 if (tmp->nid_brw_stats == NULL)
230                         RETURN(-ENOMEM);
231
232                 init_brw_stats(tmp->nid_brw_stats);
233                 rc = lprocfs_seq_create(exp->exp_nid_stats->nid_proc, "brw_stats",
234                                         0644, &filter_per_nid_stats_fops,
235                                         exp->exp_nid_stats);
236                 if (rc)
237                         CWARN("Error adding the brw_stats file\n");
238
239                 rc = lprocfs_init_rw_stats(obd, &exp->exp_nid_stats->nid_stats);
240                 if (rc)
241                         RETURN(rc);
242
243                 rc = lprocfs_register_stats(tmp->nid_proc, "stats",
244                                             tmp->nid_stats);
245                 if (rc)
246                         RETURN(rc);
247                 /* Always add in ldlm_stats */
248                 tmp->nid_ldlm_stats = lprocfs_alloc_stats(LDLM_LAST_OPC -
249                                                           LDLM_FIRST_OPC, 0);
250                 if (tmp->nid_ldlm_stats == NULL)
251                         return -ENOMEM;
252
253                 lprocfs_init_ldlm_stats(tmp->nid_ldlm_stats);
254                 rc = lprocfs_register_stats(tmp->nid_proc, "ldlm_stats",
255                                             tmp->nid_ldlm_stats);
256                 if (rc)
257                         RETURN(rc);
258         }
259
260         RETURN(0);
261 }
262
263 /* Add client data to the FILTER.  We use a bitmap to locate a free space
264  * in the last_rcvd file if cl_idx is -1 (i.e. a new client).
265  * Otherwise, we have just read the data from the last_rcvd file and
266  * we know its offset. */
267 static int filter_client_add(struct obd_device *obd, struct obd_export *exp,
268                              int cl_idx)
269 {
270         struct filter_obd *filter = &obd->u.filter;
271         struct filter_export_data *fed = &exp->exp_filter_data;
272         unsigned long *bitmap = filter->fo_last_rcvd_slots;
273         int new_client = (cl_idx == -1);
274
275         ENTRY;
276
277         LASSERT(bitmap != NULL);
278         LASSERTF(cl_idx > -2, "%d\n", cl_idx);
279
280         /* Self-export */
281         if (strcmp(fed->fed_lcd->lcd_uuid, obd->obd_uuid.uuid) == 0)
282                 RETURN(0);
283
284         /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
285          * there's no need for extra complication here
286          */
287         if (new_client) {
288                 cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
289         repeat:
290                 if (cl_idx >= LR_MAX_CLIENTS) {
291                         CERROR("no room for %u client - fix LR_MAX_CLIENTS\n",
292                                cl_idx);
293                         RETURN(-EOVERFLOW);
294                 }
295                 if (test_and_set_bit(cl_idx, bitmap)) {
296                         cl_idx = find_next_zero_bit(bitmap, LR_MAX_CLIENTS,
297                                                     cl_idx);
298                         goto repeat;
299                 }
300         } else {
301                 if (test_and_set_bit(cl_idx, bitmap)) {
302                         CERROR("FILTER client %d: bit already set in bitmap!\n",
303                                cl_idx);
304                         LBUG();
305                 }
306         }
307
308         fed->fed_lr_idx = cl_idx;
309         fed->fed_lr_off = le32_to_cpu(filter->fo_fsd->lsd_client_start) +
310                 cl_idx * le16_to_cpu(filter->fo_fsd->lsd_client_size);
311         LASSERTF(fed->fed_lr_off > 0, "fed_lr_off = %llu\n", fed->fed_lr_off);
312
313         CDEBUG(D_INFO, "client at index %d (%llu) with UUID '%s' added\n",
314                fed->fed_lr_idx, fed->fed_lr_off, fed->fed_lcd->lcd_uuid);
315
316         if (new_client) {
317                 struct lvfs_run_ctxt saved;
318                 loff_t off = fed->fed_lr_off;
319                 int rc;
320                 void *handle;
321
322                 CDEBUG(D_INFO, "writing client lcd at idx %u (%llu) (len %u)\n",
323                        fed->fed_lr_idx,off,(unsigned int)sizeof(*fed->fed_lcd));
324
325                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
326                 /* Transaction needed to fix bug 1403 */
327                 handle = fsfilt_start(obd,
328                                       filter->fo_rcvd_filp->f_dentry->d_inode,
329                                       FSFILT_OP_SETATTR, NULL);
330                 if (IS_ERR(handle)) {
331                         rc = PTR_ERR(handle);
332                         CERROR("unable to start transaction: rc %d\n", rc);
333                 } else {
334                         rc = fsfilt_add_journal_cb(obd, 0, handle,
335                                                    target_client_add_cb, exp);
336                         if (rc == 0) {
337                                 spin_lock(&exp->exp_lock);
338                                 exp->exp_need_sync = 1;
339                                 spin_unlock(&exp->exp_lock);
340                         }
341                         rc = fsfilt_write_record(obd, filter->fo_rcvd_filp,
342                                                  fed->fed_lcd,
343                                                  sizeof(*fed->fed_lcd),
344                                                  &off, rc /* sync if no cb */);
345                         fsfilt_commit(obd,
346                                       filter->fo_rcvd_filp->f_dentry->d_inode,
347                                       handle, 0);
348                 }
349                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
350
351                 if (rc) {
352                         CERROR("error writing %s client idx %u: rc %d\n",
353                                LAST_RCVD, fed->fed_lr_idx, rc);
354                         RETURN(rc);
355                 }
356         }
357         RETURN(0);
358 }
359
360 static int filter_client_free(struct obd_export *exp)
361 {
362         struct filter_export_data *fed = &exp->exp_filter_data;
363         struct filter_obd *filter = &exp->exp_obd->u.filter;
364         struct obd_device *obd = exp->exp_obd;
365         struct lsd_client_data zero_lcd;
366         struct lvfs_run_ctxt saved;
367         int rc;
368         loff_t off;
369         ENTRY;
370
371         if (fed->fed_lcd == NULL)
372                 RETURN(0);
373
374         /* XXX if lcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
375         if (strcmp(fed->fed_lcd->lcd_uuid, obd->obd_uuid.uuid ) == 0)
376                 GOTO(free, 0);
377
378         CDEBUG(D_INFO, "freeing client at idx %u, offset %lld with UUID '%s'\n",
379                fed->fed_lr_idx, fed->fed_lr_off, fed->fed_lcd->lcd_uuid);
380
381         LASSERT(filter->fo_last_rcvd_slots != NULL);
382
383         off = fed->fed_lr_off;
384
385         /* Don't clear fed_lr_idx here as it is likely also unset.  At worst
386          * we leak a client slot that will be cleaned on the next recovery. */
387         if (off <= 0) {
388                 CERROR("%s: client idx %d has med_off %lld\n",
389                        obd->obd_name, fed->fed_lr_idx, off);
390                 GOTO(free, rc = -EINVAL);
391         }
392
393         /* Clear the bit _after_ zeroing out the client so we don't
394            race with filter_client_add and zero out new clients.*/
395         if (!test_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
396                 CERROR("FILTER client %u: bit already clear in bitmap!!\n",
397                        fed->fed_lr_idx);
398                 LBUG();
399         }
400
401         if (!(exp->exp_flags & OBD_OPT_FAILOVER)) {
402                 memset(&zero_lcd, 0, sizeof zero_lcd);
403                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
404                 rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, &zero_lcd,
405                                          sizeof(zero_lcd), &off,
406                                          (!exp->exp_libclient ||
407                                           exp->exp_need_sync));
408                 if (rc == 0)
409                         /* update server's transno */
410                         filter_update_server_data(obd, filter->fo_rcvd_filp,
411                                                   filter->fo_fsd,
412                                                   !exp->exp_libclient);
413                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
414
415                 CDEBUG(rc == 0 ? D_INFO : D_ERROR,
416                        "zeroing out client %s at idx %u (%llu) in %s rc %d\n",
417                        fed->fed_lcd->lcd_uuid, fed->fed_lr_idx, fed->fed_lr_off,
418                        LAST_RCVD, rc);
419         }
420
421         if (!test_and_clear_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
422                 CERROR("FILTER client %u: bit already clear in bitmap!!\n",
423                        fed->fed_lr_idx);
424                 LBUG();
425         }
426
427         EXIT;
428 free:
429         OBD_FREE_PTR(fed->fed_lcd);
430         fed->fed_lcd = NULL;
431
432         return 0;
433 }
434
435 /* drop fmd reference, free it if last ref. must be called with fed_lock held.*/
436 static inline void filter_fmd_put_nolock(struct filter_export_data *fed,
437                                          struct filter_mod_data *fmd)
438 {
439         LASSERT_SPIN_LOCKED(&fed->fed_lock);
440         if (--fmd->fmd_refcount == 0) {
441                 /* XXX when we have persistent reservations and the handle
442                  * is stored herein we need to drop it here. */
443                 fed->fed_mod_count--;
444                 list_del(&fmd->fmd_list);
445                 OBD_SLAB_FREE(fmd, ll_fmd_cachep, sizeof(*fmd));
446         }
447 }
448
449 /* drop fmd reference, free it if last ref */
450 void filter_fmd_put(struct obd_export *exp, struct filter_mod_data *fmd)
451 {
452         struct filter_export_data *fed;
453
454         if (fmd == NULL)
455                 return;
456
457         fed = &exp->exp_filter_data;
458         spin_lock(&fed->fed_lock);
459         filter_fmd_put_nolock(fed, fmd); /* caller reference */
460         spin_unlock(&fed->fed_lock);
461 }
462
463 /* expire entries from the end of the list if there are too many
464  * or they are too old */
465 static void filter_fmd_expire_nolock(struct filter_obd *filter,
466                                      struct filter_export_data *fed,
467                                      struct filter_mod_data *keep)
468 {
469         struct filter_mod_data *fmd, *tmp;
470
471         list_for_each_entry_safe(fmd, tmp, &fed->fed_mod_list, fmd_list) {
472                 if (fmd == keep)
473                         break;
474
475                 if (time_before(jiffies, fmd->fmd_expire) &&
476                     fed->fed_mod_count < filter->fo_fmd_max_num)
477                         break;
478
479                 list_del_init(&fmd->fmd_list);
480                 filter_fmd_put_nolock(fed, fmd); /* list reference */
481         }
482 }
483
484 void filter_fmd_expire(struct obd_export *exp)
485 {
486         spin_lock(&exp->exp_filter_data.fed_lock);
487         filter_fmd_expire_nolock(&exp->exp_obd->u.filter,
488                                  &exp->exp_filter_data, NULL);
489         spin_unlock(&exp->exp_filter_data.fed_lock);
490 }
491
492 /* find specified objid, group in export fmd list.
493  * caller must hold fed_lock and take fmd reference itself */
494 static struct filter_mod_data *filter_fmd_find_nolock(struct filter_obd *filter,
495                                                 struct filter_export_data *fed,
496                                                 obd_id objid, obd_gr group)
497 {
498         struct filter_mod_data *found = NULL, *fmd;
499
500         LASSERT_SPIN_LOCKED(&fed->fed_lock);
501
502         list_for_each_entry_reverse(fmd, &fed->fed_mod_list, fmd_list) {
503                 if (fmd->fmd_id == objid && fmd->fmd_gr == group) {
504                         found = fmd;
505                         list_del(&fmd->fmd_list);
506                         list_add_tail(&fmd->fmd_list, &fed->fed_mod_list);
507                         fmd->fmd_expire = jiffies + filter->fo_fmd_max_age;
508                         break;
509                 }
510         }
511
512         filter_fmd_expire_nolock(filter, fed, found);
513
514         return found;
515 }
516
517 /* Find fmd based on objid and group, or return NULL if not found. */
518 struct filter_mod_data *filter_fmd_find(struct obd_export *exp,
519                                         obd_id objid, obd_gr group)
520 {
521         struct filter_mod_data *fmd;
522
523         spin_lock(&exp->exp_filter_data.fed_lock);
524         fmd = filter_fmd_find_nolock(&exp->exp_obd->u.filter,
525                                      &exp->exp_filter_data, objid, group);
526         if (fmd)
527                 fmd->fmd_refcount++;    /* caller reference */
528         spin_unlock(&exp->exp_filter_data.fed_lock);
529
530         return fmd;
531 }
532
533 /* Find fmd based on objid and group, or create a new one if none is found.
534  * It is possible for this function to return NULL under memory pressure,
535  * or if objid = 0 is passed (which will only cause old entries to expire).
536  * Currently this is not fatal because any fmd state is transient and
537  * may also be freed when it gets sufficiently old. */
538 struct filter_mod_data *filter_fmd_get(struct obd_export *exp,
539                                        obd_id objid, obd_gr group)
540 {
541         struct filter_export_data *fed = &exp->exp_filter_data;
542         struct filter_mod_data *found = NULL, *fmd_new = NULL;
543
544         OBD_SLAB_ALLOC(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO, sizeof(*fmd_new));
545
546         spin_lock(&fed->fed_lock);
547         found = filter_fmd_find_nolock(&exp->exp_obd->u.filter,fed,objid,group);
548         if (fmd_new) {
549                 if (found == NULL) {
550                         list_add_tail(&fmd_new->fmd_list, &fed->fed_mod_list);
551                         fmd_new->fmd_id = objid;
552                         fmd_new->fmd_gr = group;
553                         fmd_new->fmd_refcount++;   /* list reference */
554                         found = fmd_new;
555                         fed->fed_mod_count++;
556                 } else {
557                         OBD_SLAB_FREE(fmd_new, ll_fmd_cachep, sizeof(*fmd_new));
558                 }
559         }
560         if (found) {
561                 found->fmd_refcount++;          /* caller reference */
562                 found->fmd_expire = jiffies +
563                         exp->exp_obd->u.filter.fo_fmd_max_age;
564         }
565
566         spin_unlock(&fed->fed_lock);
567
568         return found;
569 }
570
571 #ifdef DO_FMD_DROP
572 /* drop fmd list reference so it will disappear when last reference is put.
573  * This isn't so critical because it would in fact only affect the one client
574  * that is doing the unlink and at worst we have an stale entry referencing
575  * an object that should never be used again. */
576 static void filter_fmd_drop(struct obd_export *exp, obd_id objid, obd_gr group)
577 {
578         struct filter_mod_data *found = NULL;
579
580         spin_lock(&exp->exp_filter_data.fed_lock);
581         found = filter_fmd_find_nolock(&exp->exp_filter_data, objid, group);
582         if (found) {
583                 list_del_init(&found->fmd_list);
584                 filter_fmd_put_nolock(&exp->exp_filter_data, found);
585         }
586         spin_unlock(&exp->exp_filter_data.fed_lock);
587 }
588 #else
589 #define filter_fmd_drop(exp, objid, group)
590 #endif
591
592 /* remove all entries from fmd list */
593 static void filter_fmd_cleanup(struct obd_export *exp)
594 {
595         struct filter_export_data *fed = &exp->exp_filter_data;
596         struct filter_mod_data *fmd = NULL, *tmp;
597
598         spin_lock(&fed->fed_lock);
599         list_for_each_entry_safe(fmd, tmp, &fed->fed_mod_list, fmd_list) {
600                 list_del_init(&fmd->fmd_list);
601                 filter_fmd_put_nolock(fed, fmd);
602         }
603         spin_unlock(&fed->fed_lock);
604 }
605
606 static int filter_init_export(struct obd_export *exp)
607 {
608         spin_lock_init(&exp->exp_filter_data.fed_lock);
609         CFS_INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
610
611         spin_lock(&exp->exp_lock);
612         exp->exp_connecting = 1;
613         spin_unlock(&exp->exp_lock);
614
615         return ldlm_init_export(exp);
616 }
617
618 static int filter_free_server_data(struct filter_obd *filter)
619 {
620         OBD_FREE(filter->fo_fsd, sizeof(*filter->fo_fsd));
621         filter->fo_fsd = NULL;
622         OBD_FREE(filter->fo_last_rcvd_slots, LR_MAX_CLIENTS / 8);
623         filter->fo_last_rcvd_slots = NULL;
624         return 0;
625 }
626
627 /* assumes caller is already in kernel ctxt */
628 int filter_update_server_data(struct obd_device *obd, struct file *filp,
629                               struct lr_server_data *fsd, int force_sync)
630 {
631         loff_t off = 0;
632         int rc;
633         ENTRY;
634
635         CDEBUG(D_INODE, "server uuid      : %s\n", fsd->lsd_uuid);
636         CDEBUG(D_INODE, "server last_rcvd : "LPU64"\n",
637                le64_to_cpu(fsd->lsd_last_transno));
638         CDEBUG(D_INODE, "server last_mount: "LPU64"\n",
639                le64_to_cpu(fsd->lsd_mount_count));
640
641         fsd->lsd_compat14 = fsd->lsd_last_transno;
642         rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off, force_sync);
643         if (rc)
644                 CERROR("error writing lr_server_data: rc = %d\n", rc);
645
646         RETURN(rc);
647 }
648
649 int filter_update_last_objid(struct obd_device *obd, obd_gr group,
650                              int force_sync)
651 {
652         struct filter_obd *filter = &obd->u.filter;
653         __u64 tmp;
654         loff_t off = 0;
655         int rc;
656         ENTRY;
657
658         if (filter->fo_last_objid_files[group] == NULL) {
659                 CERROR("Object group "LPU64" not fully setup; not updating "
660                        "last_objid\n", group);
661                 RETURN(-EINVAL);
662         }
663
664         CDEBUG(D_INODE, "%s: server last_objid for group "LPU64": "LPU64"\n",
665                obd->obd_name, group, filter->fo_last_objids[group]);
666
667         tmp = cpu_to_le64(filter->fo_last_objids[group]);
668         rc = fsfilt_write_record(obd, filter->fo_last_objid_files[group],
669                                  &tmp, sizeof(tmp), &off, force_sync);
670         if (rc)
671                 CERROR("error writing group "LPU64" last objid: rc = %d\n",
672                        group, rc);
673         RETURN(rc);
674 }
675 extern int ost_handle(struct ptlrpc_request *req);
676 /* assumes caller has already in kernel ctxt */
677 static int filter_init_server_data(struct obd_device *obd, struct file * filp)
678 {
679         struct filter_obd *filter = &obd->u.filter;
680         struct lr_server_data *fsd;
681         struct lsd_client_data *lcd = NULL;
682         struct inode *inode = filp->f_dentry->d_inode;
683         unsigned long last_rcvd_size = i_size_read(inode);
684         __u64 mount_count;
685         int cl_idx;
686         loff_t off = 0;
687         int rc;
688
689         /* ensure padding in the struct is the correct size */
690         CLASSERT (offsetof(struct lr_server_data, lsd_padding) +
691                  sizeof(fsd->lsd_padding) == LR_SERVER_SIZE);
692         CLASSERT (offsetof(struct lsd_client_data, lcd_padding) +
693                  sizeof(lcd->lcd_padding) == LR_CLIENT_SIZE);
694
695         OBD_ALLOC(fsd, sizeof(*fsd));
696         if (!fsd)
697                 RETURN(-ENOMEM);
698         filter->fo_fsd = fsd;
699
700         OBD_ALLOC(filter->fo_last_rcvd_slots, LR_MAX_CLIENTS / 8);
701         if (filter->fo_last_rcvd_slots == NULL) {
702                 OBD_FREE(fsd, sizeof(*fsd));
703                 RETURN(-ENOMEM);
704         }
705
706         if (last_rcvd_size == 0) {
707                 LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name);
708
709                 memcpy(fsd->lsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->lsd_uuid));
710                 fsd->lsd_last_transno = 0;
711                 mount_count = fsd->lsd_mount_count = 0;
712                 fsd->lsd_server_size = cpu_to_le32(LR_SERVER_SIZE);
713                 fsd->lsd_client_start = cpu_to_le32(LR_CLIENT_START);
714                 fsd->lsd_client_size = cpu_to_le16(LR_CLIENT_SIZE);
715                 fsd->lsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT);
716                 filter->fo_subdir_count = FILTER_SUBDIR_COUNT;
717                 fsd->lsd_feature_incompat = cpu_to_le32(OBD_INCOMPAT_OST);
718         } else {
719                 rc = fsfilt_read_record(obd, filp, fsd, sizeof(*fsd), &off);
720                 if (rc) {
721                         CDEBUG(D_INODE,"OBD filter: error reading %s: rc %d\n",
722                                LAST_RCVD, rc);
723                         GOTO(err_fsd, rc);
724                 }
725                 if (strcmp(fsd->lsd_uuid, obd->obd_uuid.uuid) != 0) {
726                         LCONSOLE_ERROR_MSG(0x134, "Trying to start OBD %s "
727                                            "using the wrong disk %s. Were the "
728                                            "/dev/ assignments rearranged?\n",
729                                            obd->obd_uuid.uuid, fsd->lsd_uuid);
730                         GOTO(err_fsd, rc = -EINVAL);
731                 }
732                 mount_count = le64_to_cpu(fsd->lsd_mount_count);
733                 filter->fo_subdir_count = le16_to_cpu(fsd->lsd_subdir_count);
734                 /* COMPAT_146 */
735                 /* Assume old last_rcvd format unless I_C_LR is set */
736                 if (!(fsd->lsd_feature_incompat &
737                       cpu_to_le32(OBD_INCOMPAT_COMMON_LR)))
738                         fsd->lsd_last_transno = fsd->lsd_compat14;
739                 /* end COMPAT_146 */
740         }
741
742         if (fsd->lsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) {
743                 CERROR("%s: unsupported incompat filesystem feature(s) %x\n",
744                        obd->obd_name, le32_to_cpu(fsd->lsd_feature_incompat) &
745                        ~FILTER_INCOMPAT_SUPP);
746                 GOTO(err_fsd, rc = -EINVAL);
747         }
748         if (fsd->lsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) {
749                 CERROR("%s: unsupported read-only filesystem feature(s) %x\n",
750                        obd->obd_name, le32_to_cpu(fsd->lsd_feature_rocompat) &
751                        ~FILTER_ROCOMPAT_SUPP);
752                 /* Do something like remount filesystem read-only */
753                 GOTO(err_fsd, rc = -EINVAL);
754         }
755
756         CDEBUG(D_INODE, "%s: server last_transno : "LPU64"\n",
757                obd->obd_name, le64_to_cpu(fsd->lsd_last_transno));
758         CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n",
759                obd->obd_name, mount_count + 1);
760         CDEBUG(D_INODE, "%s: server data size: %u\n",
761                obd->obd_name, le32_to_cpu(fsd->lsd_server_size));
762         CDEBUG(D_INODE, "%s: per-client data start: %u\n",
763                obd->obd_name, le32_to_cpu(fsd->lsd_client_start));
764         CDEBUG(D_INODE, "%s: per-client data size: %u\n",
765                obd->obd_name, le32_to_cpu(fsd->lsd_client_size));
766         CDEBUG(D_INODE, "%s: server subdir_count: %u\n",
767                obd->obd_name, le16_to_cpu(fsd->lsd_subdir_count));
768         CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name,
769                last_rcvd_size <= le32_to_cpu(fsd->lsd_client_start) ? 0 :
770                (last_rcvd_size - le32_to_cpu(fsd->lsd_client_start)) /
771                 le16_to_cpu(fsd->lsd_client_size));
772
773         if (!obd->obd_replayable) {
774                 CWARN("%s: recovery support OFF\n", obd->obd_name);
775                 GOTO(out, rc = 0);
776         }
777
778         for (cl_idx = 0, off = le32_to_cpu(fsd->lsd_client_start);
779              off < last_rcvd_size; cl_idx++) {
780                 __u64 last_rcvd;
781                 struct obd_export *exp;
782                 struct filter_export_data *fed;
783
784                 if (!lcd) {
785                         OBD_ALLOC_PTR(lcd);
786                         if (!lcd)
787                                 GOTO(err_client, rc = -ENOMEM);
788                 }
789
790                 /* Don't assume off is incremented properly by
791                  * fsfilt_read_record(), in case sizeof(*lcd)
792                  * isn't the same as fsd->lsd_client_size.  */
793                 off = le32_to_cpu(fsd->lsd_client_start) +
794                         cl_idx * le16_to_cpu(fsd->lsd_client_size);
795                 rc = fsfilt_read_record(obd, filp, lcd, sizeof(*lcd), &off);
796                 if (rc) {
797                         CERROR("error reading FILT %s idx %d off %llu: rc %d\n",
798                                LAST_RCVD, cl_idx, off, rc);
799                         break; /* read error shouldn't cause startup to fail */
800                 }
801
802                 if (lcd->lcd_uuid[0] == '\0') {
803                         CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
804                                cl_idx);
805                         continue;
806                 }
807
808                 last_rcvd = le64_to_cpu(lcd->lcd_last_transno);
809
810                 /* These exports are cleaned up by filter_disconnect(), so they
811                  * need to be set up like real exports as filter_connect() does.
812                  */
813                 exp = class_new_export(obd, (struct obd_uuid *)lcd->lcd_uuid);
814
815                 CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64
816                        " srv lr: "LPU64"\n", lcd->lcd_uuid, cl_idx,
817                        last_rcvd, le64_to_cpu(fsd->lsd_last_transno));
818                 if (IS_ERR(exp)) {
819                         if (PTR_ERR(exp) == -EALREADY) {
820                                 /* export already exists, zero out this one */
821                                 CERROR("Zeroing out duplicate export due to "
822                                        "bug 10479.\n");
823                                 lcd->lcd_uuid[0] = '\0';
824                         } else {
825                                 GOTO(err_client, rc = PTR_ERR(exp));
826                         }
827                 } else {
828                         fed = &exp->exp_filter_data;
829                         fed->fed_lcd = lcd;
830                         fed->fed_group = 0; /* will be assigned at connect */
831                         filter_export_stats_init(obd, exp, NULL);
832                         rc = filter_client_add(obd, exp, cl_idx);
833                         /* can't fail for existing client */
834                         LASSERTF(rc == 0, "rc = %d\n", rc);
835
836                         lcd = NULL;
837                         spin_lock(&exp->exp_lock);
838                         exp->exp_connecting = 0;
839                         exp->exp_in_recovery = 0;
840                         spin_unlock(&exp->exp_lock);
841                         obd->obd_max_recoverable_clients++;
842                         class_export_put(exp);
843                 }
844
845                 /* Need to check last_rcvd even for duplicated exports. */
846                 CDEBUG(D_OTHER, "client at idx %d has last_rcvd = "LPU64"\n",
847                        cl_idx, last_rcvd);
848
849                 if (last_rcvd > le64_to_cpu(fsd->lsd_last_transno))
850                         fsd->lsd_last_transno = cpu_to_le64(last_rcvd);
851         }
852
853         if (lcd)
854                 OBD_FREE_PTR(lcd);
855
856         obd->obd_last_committed = le64_to_cpu(fsd->lsd_last_transno);
857
858         target_recovery_init(obd, ost_handle);
859
860 out:
861         filter->fo_mount_count = mount_count + 1;
862         fsd->lsd_mount_count = cpu_to_le64(filter->fo_mount_count);
863
864         /* save it, so mount count and last_transno is current */
865         rc = filter_update_server_data(obd, filp, filter->fo_fsd, 1);
866         if (rc)
867                 GOTO(err_client, rc);
868
869         RETURN(0);
870
871 err_client:
872         target_recovery_fini(obd);
873 err_fsd:
874         filter_free_server_data(filter);
875         RETURN(rc);
876 }
877
878 static int filter_cleanup_groups(struct obd_device *obd)
879 {
880         struct filter_obd *filter = &obd->u.filter;
881         struct file *filp;
882         struct dentry *dentry;
883         int i, j;
884         ENTRY;
885
886         if (filter->fo_dentry_O_groups != NULL) {
887                 for (i = 0; i < filter->fo_group_count; i++) {
888                         dentry = filter->fo_dentry_O_groups[i];
889                         if (dentry != NULL)
890                                 f_dput(dentry);
891                 }
892                 OBD_FREE(filter->fo_dentry_O_groups,
893                          filter->fo_group_count *
894                          sizeof(*filter->fo_dentry_O_groups));
895                 filter->fo_dentry_O_groups = NULL;
896         }
897         if (filter->fo_last_objid_files != NULL) {
898                 for (i = 0; i < filter->fo_group_count; i++) {
899                         filp = filter->fo_last_objid_files[i];
900                         if (filp != NULL)
901                                 filp_close(filp, 0);
902                 }
903                 OBD_FREE(filter->fo_last_objid_files,
904                          filter->fo_group_count *
905                          sizeof(*filter->fo_last_objid_files));
906                 filter->fo_last_objid_files = NULL;
907         }
908         if (filter->fo_dentry_O_sub != NULL) {
909                 for (i = 0; i < filter->fo_group_count; i++) {
910                         for (j = 0; j < filter->fo_subdir_count; j++) {
911                                 dentry = filter->fo_dentry_O_sub[i].dentry[j];
912                                 if (dentry != NULL)
913                                         f_dput(dentry);
914                         }
915                 }
916                 OBD_FREE(filter->fo_dentry_O_sub,
917                          filter->fo_group_count *
918                          sizeof(*filter->fo_dentry_O_sub));
919                 filter->fo_dentry_O_sub = NULL;
920         }
921         if (filter->fo_last_objids != NULL) {
922                 OBD_FREE(filter->fo_last_objids,
923                          filter->fo_group_count *
924                          sizeof(*filter->fo_last_objids));
925                 filter->fo_last_objids = NULL;
926         }
927         if (filter->fo_dentry_O != NULL) {
928                 f_dput(filter->fo_dentry_O);
929                 filter->fo_dentry_O = NULL;
930         }
931         RETURN(0);
932 }
933
934 static int filter_update_last_group(struct obd_device *obd, int group)
935 {
936         struct filter_obd *filter = &obd->u.filter;
937         struct file *filp = NULL;
938         int last_group = 0, rc;
939         loff_t off = 0;
940         ENTRY;
941
942         if (group <= filter->fo_committed_group)
943                 RETURN(0);
944
945         filp = filp_open("LAST_GROUP", O_RDWR, 0700);
946         if (IS_ERR(filp)) {
947                 rc = PTR_ERR(filp);
948                 filp = NULL;
949                 CERROR("cannot open LAST_GROUP: rc = %d\n", rc);
950                 GOTO(cleanup, rc);
951         }
952
953         rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
954         if (rc) {
955                 CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n",rc);
956                 GOTO(cleanup, rc);
957         }
958         LASSERT(off == 0 || last_group >= FILTER_MIN_GROUPS);
959         CDEBUG(D_INODE, "%s: previous %d, new %d\n",
960                obd->obd_name, last_group, group);
961
962         off = 0;
963         last_group = group;
964         /* must be sync: bXXXX */
965         rc = fsfilt_write_record(obd, filp, &last_group, sizeof(__u32), &off, 1);
966         if (rc) {
967                 CDEBUG(D_INODE, "error updating LAST_GROUP: rc %d\n", rc);
968                 GOTO(cleanup, rc);
969         }
970
971         filter->fo_committed_group = group;
972 cleanup:
973         if (filp)
974                 filp_close(filp, 0);
975         RETURN(rc);
976 }
977
978 static int filter_read_group_internal(struct obd_device *obd, int group,
979                                       int create)
980 {
981         struct filter_obd *filter = &obd->u.filter;
982         __u64 *new_objids = NULL;
983         struct filter_subdirs *new_subdirs = NULL, *tmp_subdirs = NULL;
984         struct dentry **new_groups = NULL;
985         struct file **new_files = NULL;
986         struct dentry *dentry;
987         struct file *filp;
988         int old_count = filter->fo_group_count, rc, stage = 0, i;
989         char name[25];
990         __u64 last_objid;
991         loff_t off = 0;
992         int len = group + 1;
993
994         snprintf(name, 24, "%d", group);
995         name[24] = '\0';
996
997         if (!create) {
998                 dentry = ll_lookup_one_len(name, filter->fo_dentry_O,
999                                            strlen(name));
1000                 if (IS_ERR(dentry)) {
1001                         CERROR("Cannot lookup expected object group %d: %ld\n",
1002                                group, PTR_ERR(dentry));
1003                         RETURN(PTR_ERR(dentry));
1004                 }
1005         } else {
1006                 dentry = simple_mkdir(filter->fo_dentry_O, filter->fo_vfsmnt,
1007                                       name, 0700, 1);
1008                 if (IS_ERR(dentry)) {
1009                         CERROR("cannot lookup/create O/%s: rc = %ld\n", name,
1010                                PTR_ERR(dentry));
1011                         RETURN(PTR_ERR(dentry));
1012                 }
1013         }
1014         stage = 1;
1015
1016         snprintf(name, 24, "O/%d/LAST_ID", group);
1017         name[24] = '\0';
1018         filp = filp_open(name, O_CREAT | O_RDWR, 0700);
1019         if (IS_ERR(filp)) {
1020                 CERROR("cannot create %s: rc = %ld\n", name, PTR_ERR(filp));
1021                 GOTO(cleanup, rc = PTR_ERR(filp));
1022         }
1023         stage = 2;
1024
1025         rc = fsfilt_read_record(obd, filp, &last_objid, sizeof(__u64), &off);
1026         if (rc) {
1027                 CDEBUG(D_INODE, "error reading %s: rc %d\n", name, rc);
1028                 GOTO(cleanup, rc);
1029         }
1030
1031         if (filter->fo_subdir_count) {
1032                 OBD_ALLOC(tmp_subdirs, sizeof(*tmp_subdirs));
1033                 if (tmp_subdirs == NULL)
1034                         GOTO(cleanup, rc = -ENOMEM);
1035                 stage = 3;
1036
1037                 for (i = 0; i < filter->fo_subdir_count; i++) {
1038                         char dir[20];
1039                         snprintf(dir, sizeof(dir), "d%u", i);
1040
1041                         tmp_subdirs->dentry[i] = simple_mkdir(dentry,
1042                                                               filter->fo_vfsmnt,
1043                                                               dir, 0700, 1);
1044                         if (IS_ERR(tmp_subdirs->dentry[i])) {
1045                                 rc = PTR_ERR(tmp_subdirs->dentry[i]);
1046                                 CERROR("can't lookup/create O/%d/%s: rc = %d\n",
1047                                        group, dir, rc);
1048                                 GOTO(cleanup, rc);
1049                         }
1050
1051                         CDEBUG(D_INODE, "got/created O/%d/%s: %p\n", group, dir,
1052                                tmp_subdirs->dentry[i]);
1053                 }
1054         }
1055
1056         /* 'group' is an index; we need an array of length 'group + 1' */
1057         if (group + 1 > old_count) {
1058                 OBD_ALLOC(new_objids, len * sizeof(*new_objids));
1059                 OBD_ALLOC(new_subdirs, len * sizeof(*new_subdirs));
1060                 OBD_ALLOC(new_groups, len * sizeof(*new_groups));
1061                 OBD_ALLOC(new_files, len * sizeof(*new_files));
1062                 stage = 4;
1063                 if (new_objids == NULL || new_subdirs == NULL ||
1064                     new_groups == NULL || new_files == NULL)
1065                         GOTO(cleanup, rc = -ENOMEM);
1066
1067                 if (old_count) {
1068                         memcpy(new_objids, filter->fo_last_objids,
1069                                old_count * sizeof(*new_objids));
1070                         memcpy(new_subdirs, filter->fo_dentry_O_sub,
1071                                old_count * sizeof(*new_subdirs));
1072                         memcpy(new_groups, filter->fo_dentry_O_groups,
1073                                old_count * sizeof(*new_groups));
1074                         memcpy(new_files, filter->fo_last_objid_files,
1075                                old_count * sizeof(*new_files));
1076
1077                         OBD_FREE(filter->fo_last_objids,
1078                                  old_count * sizeof(*new_objids));
1079                         OBD_FREE(filter->fo_dentry_O_sub,
1080                                  old_count * sizeof(*new_subdirs));
1081                         OBD_FREE(filter->fo_dentry_O_groups,
1082                                  old_count * sizeof(*new_groups));
1083                         OBD_FREE(filter->fo_last_objid_files,
1084                                  old_count * sizeof(*new_files));
1085                 }
1086                 filter->fo_last_objids = new_objids;
1087                 filter->fo_dentry_O_sub = new_subdirs;
1088                 filter->fo_dentry_O_groups = new_groups;
1089                 filter->fo_last_objid_files = new_files;
1090                 filter->fo_group_count = len;
1091         }
1092
1093         filter->fo_dentry_O_groups[group] = dentry;
1094         filter->fo_last_objid_files[group] = filp;
1095         if (filter->fo_subdir_count) {
1096                 filter->fo_dentry_O_sub[group] = *tmp_subdirs;
1097                 OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
1098         }
1099
1100         filter_update_last_group(obd, group);
1101
1102         if (i_size_read(filp->f_dentry->d_inode) == 0) {
1103                 filter->fo_last_objids[group] = FILTER_INIT_OBJID;
1104                 rc = filter_update_last_objid(obd, group, 1);
1105                 RETURN(rc);
1106         }
1107
1108         filter->fo_last_objids[group] = le64_to_cpu(last_objid);
1109         CDEBUG(D_INODE, "%s: server last_objid group %d: "LPU64"\n",
1110                obd->obd_name, group, last_objid);
1111         RETURN(0);
1112  cleanup:
1113         switch (stage) {
1114         case 4:
1115                 if (new_objids != NULL)
1116                         OBD_FREE(new_objids, len * sizeof(*new_objids));
1117                 if (new_subdirs != NULL)
1118                         OBD_FREE(new_subdirs, len * sizeof(*new_subdirs));
1119                 if (new_groups != NULL)
1120                         OBD_FREE(new_groups, len * sizeof(*new_groups));
1121                 if (new_files != NULL)
1122                         OBD_FREE(new_files, len * sizeof(*new_files));
1123         case 3:
1124                 if (filter->fo_subdir_count) {
1125                         for (i = 0; i < filter->fo_subdir_count; i++) {
1126                                 if (tmp_subdirs->dentry[i] != NULL)
1127                                         dput(tmp_subdirs->dentry[i]);
1128                         }
1129                         OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
1130                 }
1131         case 2:
1132                 filp_close(filp, 0);
1133         case 1:
1134                 dput(dentry);
1135         }
1136         RETURN(rc);
1137 }
1138
1139 static int filter_read_groups(struct obd_device *obd, int last_group,
1140                               int create)
1141 {
1142         struct filter_obd *filter = &obd->u.filter;
1143         int old_count, group, rc = 0;
1144
1145         down(&filter->fo_init_lock);
1146         old_count = filter->fo_group_count;
1147         for (group = old_count; group <= last_group; group++) {
1148                 if (group == 0)
1149                         continue; /* no group zero */
1150
1151                 rc = filter_read_group_internal(obd, group, create);
1152                 if (rc != 0)
1153                         break;
1154         }
1155         up(&filter->fo_init_lock);
1156         return rc;
1157 }
1158
1159 /* FIXME: object groups */
1160 static int filter_prep_groups(struct obd_device *obd)
1161 {
1162         struct filter_obd *filter = &obd->u.filter;
1163         struct dentry *dentry, *O_dentry;
1164         struct file *filp;
1165         int    last_group, rc = 0, cleanup_phase = 0;
1166         loff_t off = 0;
1167         ENTRY;
1168
1169         O_dentry = simple_mkdir(current->fs->pwd, filter->fo_vfsmnt,
1170                                 "O", 0700, 1);
1171         CDEBUG(D_INODE, "got/created O: %p\n", O_dentry);
1172         if (IS_ERR(O_dentry)) {
1173                 rc = PTR_ERR(O_dentry);
1174                 CERROR("cannot open/create O: rc = %d\n", rc);
1175                 GOTO(cleanup, rc);
1176         }
1177         filter->fo_dentry_O = O_dentry;
1178         cleanup_phase = 1; /* O_dentry */
1179
1180         /* Lookup "R" to tell if we're on an old OST FS and need to convert
1181          * from O/R/<dir>/<objid> to O/0/<dir>/<objid>.  This can be removed
1182          * some time post 1.0 when all old-style OSTs have converted along
1183          * with the init_objid hack. */
1184         dentry = ll_lookup_one_len("R", O_dentry, 1);
1185         if (IS_ERR(dentry))
1186                 GOTO(cleanup, rc = PTR_ERR(dentry));
1187         if (dentry->d_inode && S_ISDIR(dentry->d_inode->i_mode)) {
1188                 struct dentry *O0_dentry = lookup_one_len("0", O_dentry, 1);
1189                 ENTRY;
1190
1191                 CWARN("converting OST to new object layout\n");
1192                 if (IS_ERR(O0_dentry)) {
1193                         rc = PTR_ERR(O0_dentry);
1194                         CERROR("error looking up O/0: rc %d\n", rc);
1195                         GOTO(cleanup_R, rc);
1196                 }
1197
1198                 if (O0_dentry->d_inode) {
1199                         CERROR("Both O/R and O/0 exist. Fix manually.\n");
1200                         GOTO(cleanup_O0, rc = -EEXIST);
1201                 }
1202
1203                 LOCK_INODE_MUTEX(O_dentry->d_inode);
1204                 rc = ll_vfs_rename(O_dentry->d_inode, dentry, filter->fo_vfsmnt,
1205                                    O_dentry->d_inode, O0_dentry,
1206                                    filter->fo_vfsmnt);
1207                 UNLOCK_INODE_MUTEX(O_dentry->d_inode);
1208
1209                 if (rc) {
1210                         CERROR("error renaming O/R to O/0: rc %d\n", rc);
1211                         GOTO(cleanup_O0, rc);
1212                 }
1213                 filter->fo_fsd->lsd_feature_incompat |=
1214                         cpu_to_le32(OBD_INCOMPAT_GROUPS);
1215                 rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
1216                                                filter->fo_fsd, 1);
1217                 GOTO(cleanup_O0, rc);
1218
1219         cleanup_O0:
1220                 f_dput(O0_dentry);
1221         cleanup_R:
1222                 f_dput(dentry);
1223                 if (rc)
1224                         GOTO(cleanup, rc);
1225         } else {
1226                 f_dput(dentry);
1227         }
1228
1229         cleanup_phase = 2; /* groups */
1230
1231         /* we have to initialize all groups before first connections from
1232          * clients because they may send create/destroy for any group -bzzz */
1233         filp = filp_open("LAST_GROUP", O_CREAT | O_RDWR, 0700);
1234         if (IS_ERR(filp)) {
1235                 CERROR("cannot create LAST_GROUP: rc = %ld\n", PTR_ERR(filp));
1236                 GOTO(cleanup, rc = PTR_ERR(filp));
1237         }
1238         cleanup_phase = 3; /* filp */
1239
1240         rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
1241         if (rc) {
1242                 CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n", rc);
1243                 GOTO(cleanup, rc);
1244         }
1245         if (off == 0) {
1246                 last_group = FILTER_MIN_GROUPS;
1247         } else {
1248                 LASSERT(last_group >= FILTER_MIN_GROUPS);
1249         }
1250
1251         CWARN("%s: initialize groups [%d,%d]\n", obd->obd_name,
1252               FILTER_MIN_GROUPS, last_group);
1253         filter->fo_committed_group = last_group;
1254         rc = filter_read_groups(obd, last_group, 1);
1255         if (rc)
1256                 GOTO(cleanup, rc);
1257
1258         filp_close(filp, 0);
1259         RETURN(0);
1260
1261  cleanup:
1262         switch (cleanup_phase) {
1263         case 3:
1264                 filp_close(filp, 0);
1265         case 2:
1266                 filter_cleanup_groups(obd);
1267         case 1:
1268                 f_dput(filter->fo_dentry_O);
1269                 filter->fo_dentry_O = NULL;
1270         default:
1271                 break;
1272         }
1273         return rc;
1274
1275 }
1276
1277 /* setup the object store with correct subdirectories */
1278 static int filter_prep(struct obd_device *obd)
1279 {
1280         struct lvfs_run_ctxt saved;
1281         struct filter_obd *filter = &obd->u.filter;
1282         struct file *file;
1283         struct inode *inode;
1284         int rc = 0;
1285         ENTRY;
1286
1287         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1288         file = filp_open(LAST_RCVD, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
1289         if (!file || IS_ERR(file)) {
1290                 rc = PTR_ERR(file);
1291                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
1292                        LAST_RCVD, rc);
1293                 GOTO(out, rc);
1294         }
1295         filter->fo_rcvd_filp = file;
1296         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
1297                 CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD,
1298                        file->f_dentry->d_inode->i_mode);
1299                 GOTO(err_filp, rc = -ENOENT);
1300         }
1301
1302         inode = file->f_dentry->d_parent->d_inode;
1303         /* We use i_op->unlink directly in filter_vfs_unlink() */
1304         if (!inode->i_op || !inode->i_op->create || !inode->i_op->unlink) {
1305                 CERROR("%s: filesystem does not support create/unlink ops\n",
1306                        obd->obd_name);
1307                 GOTO(err_filp, rc = -EOPNOTSUPP);
1308         }
1309
1310         rc = filter_init_server_data(obd, file);
1311         if (rc) {
1312                 CERROR("cannot read %s: rc = %d\n", LAST_RCVD, rc);
1313                 GOTO(err_filp, rc);
1314         }
1315         /* open and create health check io file*/
1316         file = filp_open(HEALTH_CHECK, O_RDWR | O_CREAT, 0644);
1317         if (IS_ERR(file)) {
1318                 rc = PTR_ERR(file);
1319                 CERROR("OBD filter: cannot open/create %s rc = %d\n",
1320                        HEALTH_CHECK, rc);
1321                 GOTO(err_server_data, rc);
1322         }
1323         filter->fo_health_check_filp = file;
1324         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
1325                 CERROR("%s is not a regular file!: mode = %o\n", HEALTH_CHECK,
1326                        file->f_dentry->d_inode->i_mode);
1327                 GOTO(err_health_check, rc = -ENOENT);
1328         }
1329         rc = lvfs_check_io_health(obd, file);
1330         if (rc)
1331                 GOTO(err_health_check, rc);
1332
1333         rc = filter_prep_groups(obd);
1334         if (rc)
1335                 GOTO(err_health_check, rc);
1336 out:
1337         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1338
1339         return(rc);
1340
1341 err_health_check:
1342         if (filp_close(filter->fo_health_check_filp, 0))
1343                 CERROR("can't close %s after error\n", HEALTH_CHECK);
1344         filter->fo_health_check_filp = NULL;
1345 err_server_data:
1346         target_recovery_fini(obd);
1347         filter_free_server_data(filter);
1348 err_filp:
1349         if (filp_close(filter->fo_rcvd_filp, 0))
1350                 CERROR("can't close %s after error\n", LAST_RCVD);
1351         filter->fo_rcvd_filp = NULL;
1352         goto out;
1353 }
1354
1355 /* cleanup the filter: write last used object id to status file */
1356 static void filter_post(struct obd_device *obd)
1357 {
1358         struct lvfs_run_ctxt saved;
1359         struct filter_obd *filter = &obd->u.filter;
1360         int rc, i;
1361
1362         /* XXX: filter_update_lastobjid used to call fsync_dev.  It might be
1363          * best to start a transaction with h_sync, because we removed this
1364          * from lastobjid */
1365
1366         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1367         rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
1368                                        filter->fo_fsd, 0);
1369         if (rc)
1370                 CERROR("error writing server data: rc = %d\n", rc);
1371
1372         for (i = 1; i < filter->fo_group_count; i++) {
1373                 rc = filter_update_last_objid(obd, i,
1374                                 (i == filter->fo_group_count - 1));
1375                 if (rc)
1376                         CERROR("error writing group %d lastobjid: rc = %d\n",
1377                                i, rc);
1378         }
1379
1380         rc = filp_close(filter->fo_rcvd_filp, 0);
1381         filter->fo_rcvd_filp = NULL;
1382         if (rc)
1383                 CERROR("error closing %s: rc = %d\n", LAST_RCVD, rc);
1384
1385         rc = filp_close(filter->fo_health_check_filp, 0);
1386         filter->fo_health_check_filp = NULL;
1387         if (rc)
1388                 CERROR("error closing %s: rc = %d\n", HEALTH_CHECK, rc);
1389
1390         filter_cleanup_groups(obd);
1391         filter_free_server_data(filter);
1392         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1393
1394         filter_free_capa_keys(filter);
1395         cleanup_capa_hash(filter->fo_capa_hash);
1396 }
1397
1398 static void filter_set_last_id(struct filter_obd *filter,
1399                                obd_id id, obd_gr group)
1400 {
1401         LASSERT(filter->fo_fsd != NULL);
1402         LASSERT(group <= filter->fo_group_count);
1403
1404         spin_lock(&filter->fo_objidlock);
1405         filter->fo_last_objids[group] = id;
1406         spin_unlock(&filter->fo_objidlock);
1407 }
1408
1409 obd_id filter_last_id(struct filter_obd *filter, obd_gr group)
1410 {
1411         obd_id id;
1412         LASSERT(filter->fo_fsd != NULL);
1413         LASSERT(group <= filter->fo_group_count);
1414
1415         /* FIXME: object groups */
1416         spin_lock(&filter->fo_objidlock);
1417         id = filter->fo_last_objids[group];
1418         spin_unlock(&filter->fo_objidlock);
1419
1420         return id;
1421 }
1422
1423 static int filter_lock_dentry(struct obd_device *obd, struct dentry *dparent)
1424 {
1425         LOCK_INODE_MUTEX_PARENT(dparent->d_inode);
1426         return 0;
1427 }
1428
1429 /* We never dget the object parent, so DON'T dput it either */
1430 struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid)
1431 {
1432         struct filter_obd *filter = &obd->u.filter;
1433         struct filter_subdirs *subdirs;
1434         LASSERT(group < filter->fo_group_count); /* FIXME: object groups */
1435
1436         if ((group > 0 && group < FILTER_GROUP_MDS0) ||
1437              filter->fo_subdir_count == 0)
1438                 return filter->fo_dentry_O_groups[group];
1439
1440         subdirs = &filter->fo_dentry_O_sub[group];
1441         return subdirs->dentry[objid & (filter->fo_subdir_count - 1)];
1442 }
1443
1444 /* We never dget the object parent, so DON'T dput it either */
1445 struct dentry *filter_parent_lock(struct obd_device *obd, obd_gr group,
1446                                   obd_id objid)
1447 {
1448         unsigned long now = jiffies;
1449         struct dentry *dparent = filter_parent(obd, group, objid);
1450         int rc;
1451
1452         if (IS_ERR(dparent))
1453                 return dparent;
1454         if (dparent == NULL)
1455                 return ERR_PTR(-ENOENT);
1456
1457         rc = filter_lock_dentry(obd, dparent);
1458         fsfilt_check_slow(obd, now, "parent lock");
1459         return rc ? ERR_PTR(rc) : dparent;
1460 }
1461
1462 /* We never dget the object parent, so DON'T dput it either */
1463 static void filter_parent_unlock(struct dentry *dparent)
1464 {
1465         UNLOCK_INODE_MUTEX(dparent->d_inode);
1466 }
1467
1468 /* How to get files, dentries, inodes from object id's.
1469  *
1470  * If dir_dentry is passed, the caller has already locked the parent
1471  * appropriately for this operation (normally a write lock).  If
1472  * dir_dentry is NULL, we do a read lock while we do the lookup to
1473  * avoid races with create/destroy and such changing the directory
1474  * internal to the filesystem code. */
1475 struct dentry *filter_fid2dentry(struct obd_device *obd,
1476                                  struct dentry *dir_dentry,
1477                                  obd_gr group, obd_id id)
1478 {
1479         struct dentry *dparent = dir_dentry;
1480         struct dentry *dchild;
1481         char name[32];
1482         int len;
1483         ENTRY;
1484
1485         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT) &&
1486             obd->u.filter.fo_destroys_in_progress == 0) {
1487                 /* don't fail lookups for orphan recovery, it causes
1488                  * later LBUGs when objects still exist during precreate */
1489                 CDEBUG(D_INFO, "*** obd_fail_loc=%x ***\n",OBD_FAIL_OST_ENOENT);
1490                 RETURN(ERR_PTR(-ENOENT));
1491         }
1492         if (id == 0) {
1493                 CERROR("fatal: invalid object id 0\n");
1494                 RETURN(ERR_PTR(-ESTALE));
1495         }
1496
1497         len = sprintf(name, LPU64, id);
1498         if (dir_dentry == NULL) {
1499                 dparent = filter_parent_lock(obd, group, id);
1500                 if (IS_ERR(dparent)) {
1501                         CERROR("%s: error getting object "LPU64":"LPU64
1502                                " parent: rc %ld\n", obd->obd_name,
1503                                id, group, PTR_ERR(dparent));
1504                         RETURN(dparent);
1505                 }
1506         }
1507         CDEBUG(D_INODE, "looking up object O/%.*s/%s\n",
1508                dparent->d_name.len, dparent->d_name.name, name);
1509         dchild = /*ll_*/lookup_one_len(name, dparent, len);
1510         if (dir_dentry == NULL)
1511                 filter_parent_unlock(dparent);
1512         if (IS_ERR(dchild)) {
1513                 CERROR("%s: child lookup error %ld\n", obd->obd_name,
1514                        PTR_ERR(dchild));
1515                 RETURN(dchild);
1516         }
1517
1518         if (dchild->d_inode != NULL && is_bad_inode(dchild->d_inode)) {
1519                 CERROR("%s: got bad object "LPU64" inode %lu\n",
1520                        obd->obd_name, id, dchild->d_inode->i_ino);
1521                 f_dput(dchild);
1522                 RETURN(ERR_PTR(-ENOENT));
1523         }
1524
1525         CDEBUG(D_INODE, "got child objid %s: %p, count = %d\n",
1526                name, dchild, atomic_read(&dchild->d_count));
1527
1528         LASSERT(atomic_read(&dchild->d_count) > 0);
1529
1530         RETURN(dchild);
1531 }
1532
1533 static int filter_prepare_destroy(struct obd_device *obd, obd_id objid,
1534                                   obd_id group)
1535 {
1536         struct lustre_handle lockh;
1537         int flags = LDLM_AST_DISCARD_DATA, rc;
1538         struct ldlm_res_id res_id;
1539         ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
1540         ENTRY;
1541
1542         osc_build_res_name(objid, group, &res_id);
1543         /* Tell the clients that the object is gone now and that they should
1544          * throw away any cached pages. */
1545         rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id, LDLM_EXTENT,
1546                                     &policy, LCK_PW, &flags, ldlm_blocking_ast,
1547                                     ldlm_completion_ast, NULL, NULL, 0, NULL,
1548                                     NULL, &lockh);
1549
1550         /* We only care about the side-effects, just drop the lock. */
1551         if (rc == ELDLM_OK)
1552                 ldlm_lock_decref(&lockh, LCK_PW);
1553
1554         RETURN(rc);
1555 }
1556
1557 /* This is vfs_unlink() without down(i_sem).  If we call regular vfs_unlink()
1558  * we have 2.6 lock ordering issues with filter_commitrw_write() as it takes
1559  * i_sem before starting a handle, while filter_destroy() + vfs_unlink do the
1560  * reverse.  Caller must take i_sem before starting the transaction and we
1561  * drop it here before the inode is removed from the dentry.  bug 4180/6984 */
1562 int filter_vfs_unlink(struct inode *dir, struct dentry *dentry,
1563                       struct vfsmount *mnt)
1564 {
1565         int rc;
1566         ENTRY;
1567
1568         /* don't need dir->i_zombie for 2.4, it is for rename/unlink of dir
1569          * itself we already hold dir->i_mutex for child create/unlink ops */
1570         LASSERT(dentry->d_inode != NULL);
1571         LASSERT(TRYLOCK_INODE_MUTEX(dir) == 0);
1572         LASSERT(TRYLOCK_INODE_MUTEX(dentry->d_inode) == 0);
1573
1574
1575         /* may_delete() */
1576         if (/*!dentry->d_inode ||*/dentry->d_parent->d_inode != dir)
1577                 GOTO(out, rc = -ENOENT);
1578
1579         rc = ll_permission(dir, MAY_WRITE | MAY_EXEC, NULL);
1580         if (rc)
1581                 GOTO(out, rc);
1582
1583         if (IS_APPEND(dir))
1584                 GOTO(out, rc = -EPERM);
1585
1586         /* check_sticky() */
1587         if ((dentry->d_inode->i_uid != current->fsuid &&
1588              !cfs_capable(CFS_CAP_FOWNER)) || IS_APPEND(dentry->d_inode) ||
1589             IS_IMMUTABLE(dentry->d_inode))
1590                 GOTO(out, rc = -EPERM);
1591
1592         /* NOTE: This might need to go outside i_mutex, though it isn't clear if
1593          *       that was done because of journal_start (which is already done
1594          *       here) or some other ordering issue. */
1595         DQUOT_INIT(dir);
1596
1597         rc = ll_security_inode_unlink(dir, dentry, mnt);
1598         if (rc)
1599                 GOTO(out, rc);
1600
1601         rc = dir->i_op->unlink(dir, dentry);
1602 out:
1603         /* need to drop i_mutex before we lose inode reference */
1604         UNLOCK_INODE_MUTEX(dentry->d_inode);
1605         if (rc == 0)
1606                 d_delete(dentry);
1607
1608         RETURN(rc);
1609 }
1610
1611 /* Caller must hold LCK_PW on parent and push us into kernel context.
1612  * Caller must hold child i_mutex, we drop it always.
1613  * Caller is also required to ensure that dchild->d_inode exists. */
1614 static int filter_destroy_internal(struct obd_device *obd, obd_id objid,
1615                                    obd_gr group, struct dentry *dparent,
1616                                    struct dentry *dchild)
1617 {
1618         struct inode *inode = dchild->d_inode;
1619         struct filter_obd *filter = &obd->u.filter;
1620         int rc;
1621
1622         if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) {
1623                 CERROR("destroying objid %.*s ino %lu nlink %lu count %d\n",
1624                        dchild->d_name.len, dchild->d_name.name, inode->i_ino,
1625                        (unsigned long)inode->i_nlink,
1626                        atomic_read(&inode->i_count));
1627         }
1628
1629         rc = filter_vfs_unlink(dparent->d_inode, dchild, filter->fo_vfsmnt);
1630         if (rc)
1631                 CERROR("error unlinking objid %.*s: rc %d\n",
1632                        dchild->d_name.len, dchild->d_name.name, rc);
1633         return(rc);
1634 }
1635
1636 struct filter_intent_args {
1637         struct ldlm_lock **victim;
1638         __u64 size;
1639         int *liblustre;
1640 };
1641
1642 static enum interval_iter filter_intent_cb(struct interval_node *n,
1643                                            void *args)
1644 {
1645         struct ldlm_interval *node = (struct ldlm_interval *)n;
1646         struct filter_intent_args *arg = (struct filter_intent_args*)args;
1647         __u64 size = arg->size;
1648         struct ldlm_lock **v = arg->victim;
1649         struct ldlm_lock *lck;
1650
1651         /* If the interval is lower than the current file size,
1652          * just break. */
1653         if (interval_high(n) <= size)
1654                 return INTERVAL_ITER_STOP;
1655
1656         list_for_each_entry(lck, &node->li_group, l_sl_policy) {
1657                 /* Don't send glimpse ASTs to liblustre clients.
1658                  * They aren't listening for them, and they do
1659                  * entirely synchronous I/O anyways. */
1660                 if (lck->l_export == NULL ||
1661                     lck->l_export->exp_libclient == 1)
1662                         continue;
1663
1664                 if (*arg->liblustre)
1665                         *arg->liblustre = 0;
1666
1667                 if (*v == NULL) {
1668                         *v = LDLM_LOCK_GET(lck);
1669                 } else if ((*v)->l_policy_data.l_extent.start <
1670                            lck->l_policy_data.l_extent.start) {
1671                         LDLM_LOCK_RELEASE(*v);
1672                         *v = LDLM_LOCK_GET(lck);
1673                 }
1674
1675                 /* the same policy group - every lock has the
1676                  * same extent, so needn't do it any more */
1677                 break;
1678         }
1679
1680         return INTERVAL_ITER_CONT;
1681 }
1682
1683 static int filter_intent_policy(struct ldlm_namespace *ns,
1684                                 struct ldlm_lock **lockp, void *req_cookie,
1685                                 ldlm_mode_t mode, int flags, void *data)
1686 {
1687         CFS_LIST_HEAD(rpc_list);
1688         struct ptlrpc_request *req = req_cookie;
1689         struct ldlm_lock *lock = *lockp, *l = NULL;
1690         struct ldlm_resource *res = lock->l_resource;
1691         ldlm_processing_policy policy;
1692         struct ost_lvb *res_lvb, *reply_lvb;
1693         struct ldlm_reply *rep;
1694         ldlm_error_t err;
1695         int idx, rc, tmpflags = 0, only_liblustre = 1;
1696         struct ldlm_interval_tree *tree;
1697         struct filter_intent_args arg;
1698         __u32 repsize[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
1699                            [DLM_LOCKREPLY_OFF]   = sizeof(*rep),
1700                            [DLM_REPLY_REC_OFF]   = sizeof(*reply_lvb) };
1701         ENTRY;
1702
1703         policy = ldlm_get_processing_policy(res);
1704         LASSERT(policy != NULL);
1705         LASSERT(req != NULL);
1706
1707         rc = lustre_pack_reply(req, 3, repsize, NULL);
1708         if (rc)
1709                 RETURN(req->rq_status = rc);
1710
1711         rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF, sizeof(*rep));
1712         LASSERT(rep != NULL);
1713
1714         reply_lvb = lustre_msg_buf(req->rq_repmsg, DLM_REPLY_REC_OFF,
1715                                    sizeof(*reply_lvb));
1716         LASSERT(reply_lvb != NULL);
1717
1718         //fixup_handle_for_resent_req(req, lock, &lockh);
1719
1720         /* Call the extent policy function to see if our request can be
1721          * granted, or is blocked.
1722          * If the OST lock has LDLM_FL_HAS_INTENT set, it means a glimpse
1723          * lock, and should not be granted if the lock will be blocked.
1724          */
1725
1726         LASSERT(ns == res->lr_namespace);
1727         lock_res(res);
1728         rc = policy(lock, &tmpflags, 0, &err, &rpc_list);
1729         check_res_locked(res);
1730
1731         /* FIXME: we should change the policy function slightly, to not make
1732          * this list at all, since we just turn around and free it */
1733         while (!list_empty(&rpc_list)) {
1734                 struct ldlm_lock *wlock =
1735                         list_entry(rpc_list.next, struct ldlm_lock, l_cp_ast);
1736                 LASSERT((lock->l_flags & LDLM_FL_AST_SENT) == 0);
1737                 LASSERT(lock->l_flags & LDLM_FL_CP_REQD);
1738                 lock->l_flags &= ~LDLM_FL_CP_REQD;
1739                 list_del_init(&wlock->l_cp_ast);
1740                 LDLM_LOCK_RELEASE(wlock);
1741         }
1742
1743         /* The lock met with no resistance; we're finished. */
1744         if (rc == LDLM_ITER_CONTINUE) {
1745                 /* do not grant locks to the liblustre clients: they cannot
1746                  * handle ASTs robustly.  We need to do this while still
1747                  * holding ns_lock to avoid the lock remaining on the res_link
1748                  * list (and potentially being added to l_pending_list by an
1749                  * AST) when we are going to drop this lock ASAP. */
1750                 if (lock->l_export->exp_libclient ||
1751                     OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_GLIMPSE, 2)) {
1752                         ldlm_resource_unlink_lock(lock);
1753                         err = ELDLM_LOCK_ABORTED;
1754                 } else {
1755                         err = ELDLM_LOCK_REPLACED;
1756                 }
1757                 unlock_res(res);
1758                 RETURN(err);
1759         }
1760
1761         /* Do not grant any lock, but instead send GL callbacks.  The extent
1762          * policy nicely created a list of all PW locks for us.  We will choose
1763          * the highest of those which are larger than the size in the LVB, if
1764          * any, and perform a glimpse callback. */
1765         res_lvb = res->lr_lvb_data;
1766         LASSERT(res_lvb != NULL);
1767         *reply_lvb = *res_lvb;
1768
1769         /*
1770          * ->ns_lock guarantees that no new locks are granted, and,
1771          * therefore, that res->lr_lvb_data cannot increase beyond the
1772          * end of already granted lock. As a result, it is safe to
1773          * check against "stale" reply_lvb->lvb_size value without
1774          * res->lr_lvb_sem.
1775          */
1776         arg.size = reply_lvb->lvb_size;
1777         arg.victim = &l;
1778         arg.liblustre = &only_liblustre;
1779         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
1780                 tree = &res->lr_itree[idx];
1781                 if (tree->lit_mode == LCK_PR)
1782                         continue;
1783
1784                 interval_iterate_reverse(tree->lit_root,
1785                                          filter_intent_cb, &arg);
1786         }
1787         unlock_res(res);
1788
1789         /* There were no PW locks beyond the size in the LVB; finished. */
1790         if (l == NULL) {
1791                 if (only_liblustre) {
1792                         /* If we discovered a liblustre client with a PW lock,
1793                          * however, the LVB may be out of date!  The LVB is
1794                          * updated only on glimpse (which we don't do for
1795                          * liblustre clients) and cancel (which the client
1796                          * obviously has not yet done).  So if it has written
1797                          * data but kept the lock, the LVB is stale and needs
1798                          * to be updated from disk.
1799                          *
1800                          * Of course, this will all disappear when we switch to
1801                          * taking liblustre locks on the OST. */
1802                         ldlm_res_lvbo_update(res, NULL, 0, 1);
1803                 }
1804                 RETURN(ELDLM_LOCK_ABORTED);
1805         }
1806
1807         /*
1808          * This check is for lock taken in filter_prepare_destroy() that does
1809          * not have l_glimpse_ast set. So the logic is: if there is a lock
1810          * with no l_glimpse_ast set, this object is being destroyed already.
1811          *
1812          * Hence, if you are grabbing DLM locks on the server, always set
1813          * non-NULL glimpse_ast (e.g., ldlm_request.c:ldlm_glimpse_ast()).
1814          */
1815         if (l->l_glimpse_ast == NULL) {
1816                 /* We are racing with unlink(); just return -ENOENT */
1817                 rep->lock_policy_res1 = -ENOENT;
1818                 goto out;
1819         }
1820
1821         LASSERTF(l->l_glimpse_ast != NULL, "l == %p", l);
1822         rc = l->l_glimpse_ast(l, NULL); /* this will update the LVB */
1823         /* Update the LVB from disk if the AST failed (this is a legal race) */
1824         /*
1825          * XXX nikita: situation when ldlm_server_glimpse_ast() failed before
1826          * sending ast is not handled. This can result in lost client writes.
1827          */
1828         if (rc != 0)
1829                 ldlm_res_lvbo_update(res, NULL, 0, 1);
1830
1831         lock_res(res);
1832         *reply_lvb = *res_lvb;
1833         unlock_res(res);
1834
1835  out:
1836         LDLM_LOCK_RELEASE(l);
1837
1838         RETURN(ELDLM_LOCK_ABORTED);
1839 }
1840
1841 /*
1842  * per-obd_device iobuf pool.
1843  *
1844  * To avoid memory deadlocks in low-memory setups, amount of dynamic
1845  * allocations in write-path has to be minimized (see bug 5137).
1846  *
1847  * Pages, niobuf_local's and niobuf_remote's are pre-allocated and attached to
1848  * OST threads (see ost_thread_{init,done}()).
1849  *
1850  * "iobuf's" used by filter cannot be attached to OST thread, however, because
1851  * at the OST layer there are only (potentially) multiple obd_device of type
1852  * unknown at the time of OST thread creation.
1853  *
1854  * Instead array of iobuf's is attached to struct filter_obd (->fo_iobuf_pool
1855  * field). This array has size OST_MAX_THREADS, so that each OST thread uses
1856  * it's very own iobuf.
1857  *
1858  * Functions below
1859  *
1860  *     filter_kiobuf_pool_init()
1861  *
1862  *     filter_kiobuf_pool_done()
1863  *
1864  *     filter_iobuf_get()
1865  *
1866  * operate on this array. They are "generic" in a sense that they don't depend
1867  * on actual type of iobuf's (the latter depending on Linux kernel version).
1868  */
1869
1870 /*
1871  * destroy pool created by filter_iobuf_pool_init
1872  */
1873 static void filter_iobuf_pool_done(struct filter_obd *filter)
1874 {
1875         struct filter_iobuf **pool;
1876         int i;
1877
1878         ENTRY;
1879
1880         pool = filter->fo_iobuf_pool;
1881         if (pool != NULL) {
1882                 for (i = 0; i < filter->fo_iobuf_count; ++ i) {
1883                         if (pool[i] != NULL)
1884                                 filter_free_iobuf(pool[i]);
1885                 }
1886                 OBD_FREE(pool, filter->fo_iobuf_count * sizeof pool[0]);
1887                 filter->fo_iobuf_pool = NULL;
1888         }
1889         EXIT;
1890 }
1891
1892 /*
1893  * pre-allocate pool of iobuf's to be used by filter_{prep,commit}rw_write().
1894  */
1895 static int filter_iobuf_pool_init(struct filter_obd *filter)
1896 {
1897         void **pool;
1898
1899         ENTRY;
1900
1901
1902         OBD_ALLOC_GFP(filter->fo_iobuf_pool, OSS_THREADS_MAX * sizeof(*pool),
1903                       GFP_KERNEL);
1904         if (filter->fo_iobuf_pool == NULL)
1905                 RETURN(-ENOMEM);
1906
1907         filter->fo_iobuf_count = OSS_THREADS_MAX;
1908
1909         RETURN(0);
1910 }
1911
1912 /* Return iobuf allocated for @thread_id.  We don't know in advance how
1913  * many threads there will be so we allocate a large empty array and only
1914  * fill in those slots that are actually in use.
1915  * If we haven't allocated a pool entry for this thread before, do so now. */
1916 void *filter_iobuf_get(struct filter_obd *filter, struct obd_trans_info *oti)
1917 {
1918         int thread_id                    = oti ? oti->oti_thread_id : -1;
1919         struct filter_iobuf  *pool       = NULL;
1920         struct filter_iobuf **pool_place = NULL;
1921
1922         if (thread_id >= 0) {
1923                 LASSERT(thread_id < filter->fo_iobuf_count);
1924                 pool = *(pool_place = &filter->fo_iobuf_pool[thread_id]);
1925         }
1926
1927         if (unlikely(pool == NULL)) {
1928                 pool = filter_alloc_iobuf(filter, OBD_BRW_WRITE,
1929                                           PTLRPC_MAX_BRW_PAGES);
1930                 if (pool_place != NULL)
1931                         *pool_place = pool;
1932         }
1933
1934         return pool;
1935 }
1936
1937 /* mount the file system (secretly).  lustre_cfg parameters are:
1938  * 1 = device
1939  * 2 = fstype
1940  * 3 = flags: failover=f, failout=n
1941  * 4 = mount options
1942  */
1943 int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg,
1944                         void *option)
1945 {
1946         struct filter_obd *filter = &obd->u.filter;
1947         struct vfsmount *mnt;
1948         struct lustre_mount_info *lmi;
1949         struct obd_uuid uuid;
1950         __u8 *uuid_ptr;
1951         char *str, *label;
1952         char ns_name[48];
1953         request_queue_t *q;
1954         int rc, i;
1955         ENTRY;
1956
1957         if (lcfg->lcfg_bufcount < 3 ||
1958             LUSTRE_CFG_BUFLEN(lcfg, 1) < 1 ||
1959             LUSTRE_CFG_BUFLEN(lcfg, 2) < 1)
1960                 RETURN(-EINVAL);
1961
1962         lmi = server_get_mount(obd->obd_name);
1963         if (lmi) {
1964                 /* We already mounted in lustre_fill_super.
1965                    lcfg bufs 1, 2, 4 (device, fstype, mount opts) are ignored.*/
1966                 struct lustre_sb_info *lsi = s2lsi(lmi->lmi_sb);
1967                 mnt = lmi->lmi_mnt;
1968                 obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd));
1969         } else {
1970                 /* old path - used by lctl */
1971                 CERROR("Using old MDS mount method\n");
1972                 mnt = ll_kern_mount(lustre_cfg_string(lcfg, 2),
1973                                     MS_NOATIME|MS_NODIRATIME,
1974                                     lustre_cfg_string(lcfg, 1), option);
1975                 if (IS_ERR(mnt)) {
1976                         rc = PTR_ERR(mnt);
1977                         LCONSOLE_ERROR_MSG(0x135, "Can't mount disk %s (%d)\n",
1978                                            lustre_cfg_string(lcfg, 1), rc);
1979                         RETURN(rc);
1980                 }
1981
1982                 obd->obd_fsops = fsfilt_get_ops(lustre_cfg_string(lcfg, 2));
1983         }
1984         if (IS_ERR(obd->obd_fsops))
1985                 GOTO(err_mntput, rc = PTR_ERR(obd->obd_fsops));
1986
1987         rc = filter_iobuf_pool_init(filter);
1988         if (rc != 0)
1989                 GOTO(err_ops, rc);
1990
1991         LASSERT(!lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb)));
1992
1993         /* failover is the default */
1994         obd->obd_replayable = 1;
1995
1996         if (lcfg->lcfg_bufcount > 3 && LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
1997                 str = lustre_cfg_string(lcfg, 3);
1998                 if (strchr(str, 'n')) {
1999                         CWARN("%s: recovery disabled\n", obd->obd_name);
2000                         obd->obd_replayable = 0;
2001                 }
2002         }
2003
2004         filter->fo_vfsmnt = mnt;
2005         obd->u.obt.obt_sb = mnt->mnt_sb;
2006         filter->fo_fstype = mnt->mnt_sb->s_type->name;
2007         CDEBUG(D_SUPER, "%s: mnt = %p\n", filter->fo_fstype, mnt);
2008
2009         fsfilt_setup(obd, obd->u.obt.obt_sb);
2010
2011         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
2012         obd->obd_lvfs_ctxt.pwdmnt = mnt;
2013         obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
2014         obd->obd_lvfs_ctxt.fs = get_ds();
2015         obd->obd_lvfs_ctxt.cb_ops = filter_lvfs_ops;
2016
2017         init_mutex(&filter->fo_init_lock);
2018         filter->fo_committed_group = 0;
2019
2020         rc = filter_prep(obd);
2021         if (rc)
2022                 GOTO(err_ops, rc);
2023
2024         filter->fo_destroys_in_progress = 0;
2025         for (i = 0; i < 32; i++)
2026                 sema_init(&filter->fo_create_locks[i], 1);
2027
2028         spin_lock_init(&filter->fo_translock);
2029         spin_lock_init(&filter->fo_objidlock);
2030         CFS_INIT_LIST_HEAD(&filter->fo_export_list);
2031         sema_init(&filter->fo_alloc_lock, 1);
2032         init_brw_stats(&filter->fo_filter_stats);
2033         filter->fo_read_cache = 1; /* enable read-only cache by default */
2034         filter->fo_writethrough_cache = 1; /* disable writethrough cache */
2035         filter->fo_readcache_max_filesize = FILTER_MAX_CACHE_SIZE;
2036         filter->fo_fmd_max_num = FILTER_FMD_MAX_NUM_DEFAULT;
2037         filter->fo_fmd_max_age = FILTER_FMD_MAX_AGE_DEFAULT;
2038
2039         CFS_INIT_LIST_HEAD(&filter->fo_llog_list);
2040         spin_lock_init(&filter->fo_llog_list_lock);
2041
2042         rwlock_init(&filter->fo_sptlrpc_lock);
2043         sptlrpc_rule_set_init(&filter->fo_sptlrpc_rset);
2044
2045         filter->fo_fl_oss_capa = 0;
2046         CFS_INIT_LIST_HEAD(&filter->fo_capa_keys);
2047         filter->fo_capa_hash = init_capa_hash();
2048         if (filter->fo_capa_hash == NULL)
2049                 GOTO(err_ops, rc = -ENOMEM);
2050
2051         sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid);
2052         obd->obd_namespace = ldlm_namespace_new(obd, ns_name, LDLM_NAMESPACE_SERVER,
2053                                                 LDLM_NAMESPACE_GREEDY);
2054         if (obd->obd_namespace == NULL)
2055                 GOTO(err_post, rc = -ENOMEM);
2056         obd->obd_namespace->ns_lvbp = obd;
2057         obd->obd_namespace->ns_lvbo = &filter_lvbo;
2058         ldlm_register_intent(obd->obd_namespace, filter_intent_policy);
2059
2060         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
2061                            "filter_ldlm_cb_client", &obd->obd_ldlm_client);
2062
2063         rc = obd_llog_init(obd, &obd->obd_olg, obd, 1, NULL, NULL);
2064         if (rc) {
2065                 CERROR("failed to setup llogging subsystems\n");
2066                 GOTO(err_post, rc);
2067         }
2068
2069         rc = lquota_setup(filter_quota_interface_ref, obd);
2070         if (rc)
2071                 GOTO(err_post, rc);
2072
2073         q = bdev_get_queue(mnt->mnt_sb->s_bdev);
2074         if (q->max_sectors < q->max_hw_sectors &&
2075             q->max_sectors < PTLRPC_MAX_BRW_SIZE >> 9)
2076                 LCONSOLE_INFO("%s: underlying device %s should be tuned "
2077                               "for larger I/O requests: max_sectors = %u "
2078                               "could be up to max_hw_sectors=%u\n",
2079                               obd->obd_name, mnt->mnt_sb->s_id,
2080                               q->max_sectors, q->max_hw_sectors);
2081
2082         uuid_ptr = fsfilt_uuid(obd, obd->u.obt.obt_sb);
2083         if (uuid_ptr != NULL) {
2084                 class_uuid_unparse(uuid_ptr, &uuid);
2085                 str = uuid.uuid;
2086         } else {
2087                 str = "no UUID";
2088         }
2089
2090         label = fsfilt_get_label(obd, obd->u.obt.obt_sb);
2091
2092         if (obd->obd_recovering) {
2093                 LCONSOLE_WARN("OST %s now serving %s (%s%s%s), but will be in "
2094                               "recovery for at least %d:%.02d, or until %d "
2095                               "client%s reconnect. During this time new clients"
2096                               " will not be allowed to connect. "
2097                               "Recovery progress can be monitored by watching "
2098                               "/proc/fs/lustre/obdfilter/%s/recovery_status.\n",
2099                               obd->obd_name, lustre_cfg_string(lcfg, 1),
2100                               label ?: "", label ? "/" : "", str,
2101                               obd->obd_recovery_timeout / 60,
2102                               obd->obd_recovery_timeout % 60,
2103                               obd->obd_max_recoverable_clients,
2104                               (obd->obd_max_recoverable_clients == 1) ? "":"s",
2105                               obd->obd_name);
2106         } else {
2107                 LCONSOLE_INFO("OST %s now serving %s (%s%s%s) with recovery "
2108                               "%s\n", obd->obd_name, lustre_cfg_string(lcfg, 1),
2109                               label ?: "", label ? "/" : "", str,
2110                               obd->obd_replayable ? "enabled" : "disabled");
2111         }
2112
2113         RETURN(0);
2114
2115 err_post:
2116         filter_post(obd);
2117 err_ops:
2118         fsfilt_put_ops(obd->obd_fsops);
2119         filter_iobuf_pool_done(filter);
2120 err_mntput:
2121         server_put_mount(obd->obd_name, mnt);
2122         obd->u.obt.obt_sb = 0;
2123         return rc;
2124 }
2125
2126 static int filter_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
2127 {
2128         struct lprocfs_static_vars lvars;
2129         unsigned long addr;
2130         struct page *page;
2131         int rc;
2132
2133         CLASSERT(offsetof(struct obd_device, u.obt) ==
2134                  offsetof(struct obd_device, u.filter.fo_obt));
2135
2136         if (!LUSTRE_CFG_BUFLEN(lcfg, 1) || !LUSTRE_CFG_BUFLEN(lcfg, 2))
2137                 RETURN(-EINVAL);
2138
2139         /* 2.6.9 selinux wants a full option page for do_kern_mount (bug6471) */
2140         OBD_PAGE_ALLOC(page, CFS_ALLOC_STD);
2141         if (!page)
2142                 RETURN(-ENOMEM);
2143         addr = (unsigned long)cfs_page_address(page);
2144         clear_page((void *)addr);
2145
2146         /* lprocfs must be setup before the filter so state can be safely added
2147          * to /proc incrementally as the filter is setup */
2148         lprocfs_filter_init_vars(&lvars);
2149         if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0 &&
2150             lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST) == 0) {
2151                 /* Init obdfilter private stats here */
2152                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES,
2153                                      LPROCFS_CNTR_AVGMINMAX,
2154                                      "read_bytes", "bytes");
2155                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES,
2156                                      LPROCFS_CNTR_AVGMINMAX,
2157                                      "write_bytes", "bytes");
2158                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_GET_PAGE,
2159                                      LPROCFS_CNTR_AVGMINMAX|LPROCFS_CNTR_STDDEV,
2160                                      "get_page", "usec");
2161                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_NO_PAGE,
2162                                      LPROCFS_CNTR_AVGMINMAX,
2163                                      "get_page_failures", "num");
2164                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_ACCESS,
2165                                      LPROCFS_CNTR_AVGMINMAX,
2166                                      "cache_access", "pages");
2167                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_HIT,
2168                                      LPROCFS_CNTR_AVGMINMAX,
2169                                      "cache_hit", "pages");
2170                 lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_MISS,
2171                                      LPROCFS_CNTR_AVGMINMAX,
2172                                      "cache_miss", "pages");
2173
2174                 lproc_filter_attach_seqstat(obd);
2175                 obd->obd_proc_exports_entry = lprocfs_register("exports",
2176                                                         obd->obd_proc_entry,
2177                                                         NULL, NULL);
2178                 if (IS_ERR(obd->obd_proc_exports_entry)) {
2179                         rc = PTR_ERR(obd->obd_proc_exports_entry);
2180                         CERROR("error %d setting up lprocfs for %s\n",
2181                                rc, "exports");
2182                         obd->obd_proc_exports_entry = NULL;
2183                 }
2184         }
2185         if (obd->obd_proc_exports_entry)
2186                 lprocfs_add_simple(obd->obd_proc_exports_entry, "clear",
2187                                    lprocfs_nid_stats_clear_read,
2188                                    lprocfs_nid_stats_clear_write, obd, NULL);
2189
2190         memcpy((void *)addr, lustre_cfg_buf(lcfg, 4),
2191                LUSTRE_CFG_BUFLEN(lcfg, 4));
2192         rc = filter_common_setup(obd, lcfg, (void *)addr);
2193         OBD_PAGE_FREE(page);
2194
2195         if (rc) {
2196                 lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
2197                 lprocfs_free_per_client_stats(obd);
2198                 lprocfs_free_obd_stats(obd);
2199                 lprocfs_obd_cleanup(obd);
2200         }
2201
2202         return rc;
2203 }
2204
2205 static struct llog_operations filter_mds_ost_repl_logops;
2206
2207 static struct llog_operations filter_size_orig_logops = {
2208         .lop_setup   = llog_obd_origin_setup,
2209         .lop_cleanup = llog_obd_origin_cleanup,
2210         .lop_add     = llog_obd_origin_add
2211 };
2212
2213 static int filter_olg_fini(struct obd_llog_group *olg)
2214 {
2215         struct llog_ctxt *ctxt;
2216         int rc = 0, rc2 = 0;
2217         ENTRY;
2218
2219         ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
2220         if (ctxt)
2221                 rc = llog_cleanup(ctxt);
2222
2223         ctxt = llog_group_get_ctxt(olg, LLOG_SIZE_ORIG_CTXT);
2224         if (ctxt)
2225                 rc2 = llog_cleanup(ctxt);
2226         if (!rc)
2227                 rc = rc2;
2228
2229         RETURN(rc);
2230 }
2231
2232 static int
2233 filter_olg_init(struct obd_device *obd, struct obd_llog_group *olg,
2234                 struct obd_device *tgt)
2235 {
2236         int rc;
2237         ENTRY;
2238
2239         rc = llog_setup(obd, olg, LLOG_MDS_OST_REPL_CTXT, tgt, 0, NULL,
2240                         &filter_mds_ost_repl_logops);
2241         if (rc)
2242                 GOTO(cleanup, rc);
2243
2244         rc = llog_setup(obd, olg, LLOG_SIZE_ORIG_CTXT, tgt, 0, NULL,
2245                         &filter_size_orig_logops);
2246         if (rc)
2247                 GOTO(cleanup, rc);
2248         EXIT;
2249 cleanup:
2250         if (rc)
2251                 filter_olg_fini(olg);
2252         return rc;
2253 }
2254
2255 /**
2256  * Init the default olg, which is embeded in the obd_device, for filter.
2257  */
2258 static int
2259 filter_default_olg_init(struct obd_device *obd, struct obd_llog_group *olg,
2260                         struct obd_device *tgt)
2261 {
2262         struct filter_obd *filter = &obd->u.filter;
2263         struct llog_ctxt *ctxt;
2264         int rc;
2265         ENTRY;
2266
2267         filter->fo_lcm = llog_recov_thread_init(obd->obd_name);
2268         if (!filter->fo_lcm)
2269                 RETURN(-ENOMEM);
2270
2271         filter_mds_ost_repl_logops = llog_client_ops;
2272         filter_mds_ost_repl_logops.lop_cancel = llog_obd_repl_cancel;
2273         filter_mds_ost_repl_logops.lop_connect = llog_obd_repl_connect;
2274         filter_mds_ost_repl_logops.lop_sync = llog_obd_repl_sync;
2275
2276         rc = filter_olg_init(obd, olg, tgt);
2277         if (rc)
2278                 GOTO(cleanup_lcm, rc);
2279
2280         ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
2281         if (!ctxt) {
2282                 CERROR("Can't get ctxt for %p:%x\n", olg,
2283                        LLOG_MDS_OST_REPL_CTXT);
2284                 GOTO(cleanup_olg, rc = -ENODEV);
2285         }
2286         ctxt->loc_lcm = filter->fo_lcm;
2287         ctxt->llog_proc_cb = filter_recov_log_mds_ost_cb;
2288         llog_ctxt_put(ctxt);
2289
2290         RETURN(0);
2291 cleanup_olg:
2292         filter_olg_fini(olg);
2293 cleanup_lcm:
2294         llog_recov_thread_fini(filter->fo_lcm, 1);
2295         filter->fo_lcm = NULL;
2296         return rc;
2297 }
2298
2299 static int
2300 filter_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
2301                  struct obd_device *tgt, int count, struct llog_catid *catid,
2302                  struct obd_uuid *uuid)
2303 {
2304         struct filter_obd *filter = &obd->u.filter;
2305         struct llog_ctxt *ctxt;
2306         int rc;
2307         ENTRY;
2308
2309         LASSERT(olg != NULL);
2310         if (olg == &obd->obd_olg)
2311                 return filter_default_olg_init(obd, olg, tgt);
2312
2313         LASSERT(filter->fo_lcm != NULL);
2314         rc = filter_olg_init(obd, olg, tgt);
2315         if (rc)
2316                 RETURN(rc);
2317         ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
2318         if (!ctxt) {
2319                 CERROR("Can't get ctxt for %p:%x\n", olg,
2320                        LLOG_MDS_OST_REPL_CTXT);
2321                 filter_olg_fini(olg);
2322                 RETURN(-ENODEV);
2323         }
2324         ctxt->llog_proc_cb = filter_recov_log_mds_ost_cb;
2325         ctxt->loc_lcm = filter->fo_lcm;
2326         llog_ctxt_put(ctxt);
2327         RETURN(rc);
2328 }
2329
2330 static int filter_llog_finish(struct obd_device *obd, int count)
2331 {
2332         struct filter_obd *filter = &obd->u.filter;
2333         struct llog_ctxt *ctxt;
2334         ENTRY;
2335
2336         ctxt = llog_group_get_ctxt(&obd->obd_olg, LLOG_MDS_OST_REPL_CTXT);
2337         LASSERT(ctxt != NULL);
2338         mutex_down(&ctxt->loc_sem);
2339         if (ctxt->loc_imp) {
2340                 /*
2341                  * Balance class_import_get() in llog_receptor_accept(). This
2342                  * is safe to do here, as llog is already synchronized and its
2343                  * import may go.
2344                  */
2345                 class_import_put(ctxt->loc_imp);
2346                 ctxt->loc_imp = NULL;
2347         }
2348         mutex_up(&ctxt->loc_sem);
2349         llog_ctxt_put(ctxt);
2350
2351         if (filter->fo_lcm) {
2352                 llog_recov_thread_fini(filter->fo_lcm, obd->obd_force);
2353                 filter->fo_lcm = NULL;
2354         }
2355         RETURN(filter_olg_fini(&obd->obd_olg));
2356 }
2357
2358 /**
2359  * Find the group llog according to group index in the llog group list.
2360  */
2361 static struct obd_llog_group *
2362 filter_find_olg_internal(struct filter_obd *filter, int group)
2363 {
2364         struct obd_llog_group *olg;
2365
2366         LASSERT_SPIN_LOCKED(&filter->fo_llog_list_lock);
2367         list_for_each_entry(olg, &filter->fo_llog_list, olg_list) {
2368                 if (olg->olg_group == group)
2369                         RETURN(olg);
2370         }
2371         RETURN(NULL);
2372 }
2373
2374 /**
2375  * Find the group llog according to group index on the filter
2376  */
2377 struct obd_llog_group *filter_find_olg(struct obd_device *obd, int group)
2378 {
2379         struct obd_llog_group *olg = NULL;
2380         struct filter_obd *filter;
2381
2382         filter = &obd->u.filter;
2383
2384         if (group == FILTER_GROUP_LLOG)
2385                 RETURN(&obd->obd_olg);
2386
2387         spin_lock(&filter->fo_llog_list_lock);
2388         olg = filter_find_olg_internal(filter, group);
2389         spin_unlock(&filter->fo_llog_list_lock);
2390
2391         RETURN(olg);
2392 }
2393 /**
2394  * Find the llog_group of the filter according to the group. If it can not
2395  * find, create the llog_group, which only happens when mds is being synced
2396  * with OST.
2397  */
2398 struct obd_llog_group *filter_find_create_olg(struct obd_device *obd, int group)
2399 {
2400         struct obd_llog_group *olg = NULL;
2401         struct filter_obd *filter;
2402         int rc;
2403
2404         filter = &obd->u.filter;
2405
2406         if (group == FILTER_GROUP_LLOG)
2407                 RETURN(&obd->obd_olg);
2408
2409         spin_lock(&filter->fo_llog_list_lock);
2410         olg = filter_find_olg_internal(filter, group);
2411         if (olg) {
2412                 if (olg->olg_initializing) {
2413                         GOTO(out_unlock, olg = ERR_PTR(-EBUSY));
2414                 } else {
2415                         GOTO(out_unlock, olg);
2416                 }
2417         }
2418         OBD_ALLOC_PTR(olg);
2419         if (olg == NULL)
2420                GOTO(out_unlock, olg = ERR_PTR(-ENOMEM));
2421
2422         llog_group_init(olg, group);
2423         list_add(&olg->olg_list, &filter->fo_llog_list);
2424         olg->olg_initializing = 1;
2425         spin_unlock(&filter->fo_llog_list_lock);
2426
2427         rc = llog_cat_initialize(obd, olg, 1, NULL);
2428         if (rc) {
2429                spin_lock(&filter->fo_llog_list_lock);
2430                list_del(&olg->olg_list);
2431                spin_unlock(&filter->fo_llog_list_lock);
2432                OBD_FREE_PTR(olg);
2433                GOTO(out, olg = ERR_PTR(-ENOMEM));
2434         }
2435         spin_lock(&filter->fo_llog_list_lock);
2436         olg->olg_initializing = 0;
2437         spin_unlock(&filter->fo_llog_list_lock);
2438         CDEBUG(D_OTHER, "%s: new llog group %u (0x%p)\n",
2439               obd->obd_name, group, olg);
2440 out:
2441         RETURN(olg);
2442
2443 out_unlock:
2444         spin_unlock(&filter->fo_llog_list_lock);
2445         GOTO(out, olg);
2446 }
2447
2448 static int filter_llog_connect(struct obd_export *exp,
2449                                struct llogd_conn_body *body)
2450 {
2451         struct obd_device *obd = exp->exp_obd;
2452         struct llog_ctxt *ctxt;
2453         struct obd_llog_group *olg;
2454         int rc;
2455         ENTRY;
2456
2457         CDEBUG(D_OTHER, "%s: LLog connect for: "LPX64"/"LPX64":%x\n",
2458                obd->obd_name, body->lgdc_logid.lgl_oid,
2459                body->lgdc_logid.lgl_ogr, body->lgdc_logid.lgl_ogen);
2460
2461         olg = filter_find_olg(obd, body->lgdc_logid.lgl_ogr);
2462         if (!olg) {
2463                 CERROR(" %s: can not find olg of group %d\n",
2464                        obd->obd_name, (int)body->lgdc_logid.lgl_ogr);
2465                 RETURN(-ENOENT);
2466         }
2467         llog_group_set_export(olg, exp);
2468
2469         ctxt = llog_group_get_ctxt(olg, body->lgdc_ctxt_idx);
2470         LASSERTF(ctxt != NULL, "ctxt is not null, ctxt idx %d \n",
2471                  body->lgdc_ctxt_idx);
2472
2473         CWARN("%s: Recovery from log "LPX64"/"LPX64":%x\n",
2474               obd->obd_name, body->lgdc_logid.lgl_oid,
2475               body->lgdc_logid.lgl_ogr, body->lgdc_logid.lgl_ogen);
2476
2477         rc = llog_connect(ctxt, &body->lgdc_logid,
2478                           &body->lgdc_gen, NULL);
2479         llog_ctxt_put(ctxt);
2480         if (rc != 0)
2481                 CERROR("failed to connect rc %d idx %d\n", rc,
2482                                 body->lgdc_ctxt_idx);
2483
2484         RETURN(rc);
2485 }
2486
2487 static int filter_llog_preclean(struct obd_device *obd)
2488 {
2489         struct obd_llog_group *olg, *tmp;
2490         struct filter_obd *filter;
2491         struct list_head  remove_list;
2492         int rc = 0;
2493         ENTRY;
2494
2495         rc = obd_llog_finish(obd, 0);
2496         if (rc)
2497                 CERROR("failed to cleanup llogging subsystem\n");
2498
2499         filter = &obd->u.filter;
2500         CFS_INIT_LIST_HEAD(&remove_list);
2501
2502         spin_lock(&filter->fo_llog_list_lock);
2503         while (!list_empty(&filter->fo_llog_list)) {
2504                 olg = list_entry(filter->fo_llog_list.next,
2505                                  struct obd_llog_group, olg_list);
2506                 list_del(&olg->olg_list);
2507                 list_add(&olg->olg_list, &remove_list);
2508         }
2509         spin_unlock(&filter->fo_llog_list_lock);
2510
2511         list_for_each_entry_safe(olg, tmp, &remove_list, olg_list) {
2512                 list_del_init(&olg->olg_list);
2513                 rc = filter_olg_fini(olg);
2514                 if (rc)
2515                         CERROR("failed to cleanup llogging subsystem for %u\n",
2516                                olg->olg_group);
2517                 OBD_FREE_PTR(olg);
2518         }
2519
2520         RETURN(rc);
2521 }
2522
2523 static int filter_precleanup(struct obd_device *obd,
2524                              enum obd_cleanup_stage stage)
2525 {
2526         int rc = 0;
2527         ENTRY;
2528
2529         switch(stage) {
2530         case OBD_CLEANUP_EARLY:
2531                 break;
2532         case OBD_CLEANUP_EXPORTS:
2533                 target_cleanup_recovery(obd);
2534                 rc = filter_llog_preclean(obd);
2535                 break;
2536         }
2537         RETURN(rc);
2538 }
2539
2540 static int filter_cleanup(struct obd_device *obd)
2541 {
2542         struct filter_obd *filter = &obd->u.filter;
2543         ENTRY;
2544
2545         if (obd->obd_fail)
2546                 LCONSOLE_WARN("%s: shutting down for failover; client state "
2547                               "will be preserved.\n", obd->obd_name);
2548
2549         if (!list_empty(&obd->obd_exports)) {
2550                 CERROR("%s: still has clients!\n", obd->obd_name);
2551                 class_disconnect_exports(obd);
2552                 if (!list_empty(&obd->obd_exports)) {
2553                         CERROR("still has exports after forced cleanup?\n");
2554                         RETURN(-EBUSY);
2555                 }
2556         }
2557
2558         lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
2559         lprocfs_free_per_client_stats(obd);
2560         lprocfs_free_obd_stats(obd);
2561         lprocfs_obd_cleanup(obd);
2562         lquota_cleanup(filter_quota_interface_ref, obd);
2563
2564         /* Stop recovery before namespace cleanup. */
2565         target_stop_recovery_thread(obd);
2566         target_cleanup_recovery(obd);
2567
2568         ldlm_namespace_free(obd->obd_namespace, NULL, obd->obd_force);
2569         obd->obd_namespace = NULL;
2570
2571         sptlrpc_rule_set_free(&filter->fo_sptlrpc_rset);
2572
2573         if (obd->u.obt.obt_sb == NULL)
2574                 RETURN(0);
2575
2576         filter_post(obd);
2577
2578         LL_DQUOT_OFF(obd->u.obt.obt_sb);
2579         shrink_dcache_sb(obd->u.obt.obt_sb);
2580
2581         server_put_mount(obd->obd_name, filter->fo_vfsmnt);
2582         obd->u.obt.obt_sb = NULL;
2583
2584         fsfilt_put_ops(obd->obd_fsops);
2585
2586         filter_iobuf_pool_done(filter);
2587
2588         LCONSOLE_INFO("OST %s has stopped.\n", obd->obd_name);
2589
2590         RETURN(0);
2591 }
2592
2593 static int filter_connect_internal(struct obd_export *exp,
2594                                    struct obd_connect_data *data)
2595 {
2596         struct filter_export_data *fed = &exp->exp_filter_data;
2597
2598         if (!data)
2599                 RETURN(0);
2600
2601         CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64
2602                " ocd_version: %x ocd_grant: %d ocd_index: %u\n",
2603                exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
2604                data->ocd_connect_flags, data->ocd_version,
2605                data->ocd_grant, data->ocd_index);
2606
2607         if (fed->fed_group != 0 && fed->fed_group != data->ocd_group) {
2608                 CWARN("!!! This export (nid %s) used object group %d "
2609                        "earlier; now it's trying to use group %d!  This could "
2610                        "be a bug in the MDS. Please report to "
2611                        "http://bugzilla.lustre.org/\n",
2612                        obd_export_nid2str(exp), fed->fed_group,data->ocd_group);
2613                 RETURN(-EPROTO);
2614         }
2615         fed->fed_group = data->ocd_group;
2616
2617         data->ocd_connect_flags &= OST_CONNECT_SUPPORTED;
2618         exp->exp_connect_flags = data->ocd_connect_flags;
2619         data->ocd_version = LUSTRE_VERSION_CODE;
2620
2621         if (exp->exp_connect_flags & OBD_CONNECT_GRANT) {
2622                 obd_size left, want;
2623
2624                 spin_lock(&exp->exp_obd->obd_osfs_lock);
2625                 left = filter_grant_space_left(exp);
2626                 want = data->ocd_grant;
2627                 filter_grant(exp, fed->fed_grant, want, left);
2628                 data->ocd_grant = fed->fed_grant;
2629                 spin_unlock(&exp->exp_obd->obd_osfs_lock);
2630
2631                 CDEBUG(D_CACHE, "%s: cli %s/%p ocd_grant: %d want: "
2632                        LPU64" left: "LPU64"\n", exp->exp_obd->obd_name,
2633                        exp->exp_client_uuid.uuid, exp,
2634                        data->ocd_grant, want, left);
2635         }
2636
2637         if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
2638                 struct filter_obd *filter = &exp->exp_obd->u.filter;
2639                 struct lr_server_data *lsd = filter->fo_fsd;
2640                 int index = le32_to_cpu(lsd->lsd_ost_index);
2641
2642                 if (!(lsd->lsd_feature_compat &
2643                       cpu_to_le32(OBD_COMPAT_OST))) {
2644                         /* this will only happen on the first connect */
2645                         lsd->lsd_ost_index = cpu_to_le32(data->ocd_index);
2646                         lsd->lsd_feature_compat |= cpu_to_le32(OBD_COMPAT_OST);
2647                         filter_update_server_data(exp->exp_obd,
2648                                                   filter->fo_rcvd_filp, lsd, 1);
2649                 } else if (index != data->ocd_index) {
2650                         LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
2651                                            " %u doesn't match actual OST index"
2652                                            " %u in last_rcvd file, bad "
2653                                            "configuration?\n",
2654                                            obd_export_nid2str(exp), index,
2655                                            data->ocd_index);
2656                         RETURN(-EBADF);
2657                 }
2658         }
2659
2660         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) {
2661                 data->ocd_brw_size = 65536;
2662         } else if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
2663                 data->ocd_brw_size = min(data->ocd_brw_size,
2664                                (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT));
2665                 LASSERT(data->ocd_brw_size);
2666         }
2667
2668         if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) {
2669                 __u32 cksum_types = data->ocd_cksum_types;
2670
2671                 /* The client set in ocd_cksum_types the checksum types it
2672                  * supports. We have to mask off the algorithms that we don't
2673                  * support */
2674                 if (cksum_types & OBD_CKSUM_ALL)
2675                         data->ocd_cksum_types &= OBD_CKSUM_ALL;
2676                 else
2677                         data->ocd_cksum_types = OBD_CKSUM_CRC32;
2678
2679                 CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return "
2680                                    "%x\n", exp->exp_obd->obd_name,
2681                                    obd_export_nid2str(exp), cksum_types,
2682                                    data->ocd_cksum_types);
2683         } else {
2684                 /* This client does not support OBD_CONNECT_CKSUM
2685                  * fall back to CRC32 */
2686                 CDEBUG(D_RPCTRACE, "%s: cli %s does not support "
2687                                    "OBD_CONNECT_CKSUM, CRC32 will be used\n",
2688                                    exp->exp_obd->obd_name,
2689                                    obd_export_nid2str(exp));
2690         }
2691
2692         /* FIXME: Do the same with the MDS UUID and fsd_peeruuid.
2693          * FIXME: We don't strictly need the COMPAT flag for that,
2694          * FIXME: as fsd_peeruuid[0] will tell us if that is set.
2695          * FIXME: We needed it for the index, as index 0 is valid. */
2696
2697         RETURN(0);
2698 }
2699
2700 static int filter_reconnect(const struct lu_env *env,
2701                             struct obd_export *exp, struct obd_device *obd,
2702                             struct obd_uuid *cluuid,
2703                             struct obd_connect_data *data,
2704                             void *localdata)
2705 {
2706         int rc;
2707         ENTRY;
2708
2709         if (exp == NULL || obd == NULL || cluuid == NULL)
2710                 RETURN(-EINVAL);
2711
2712         rc = filter_connect_internal(exp, data);
2713         if (rc == 0)
2714                 filter_export_stats_init(obd, exp, localdata);
2715
2716         RETURN(rc);
2717 }
2718
2719 /* nearly identical to mds_connect */
2720 static int filter_connect(const struct lu_env *env,
2721                           struct lustre_handle *conn, struct obd_device *obd,
2722                           struct obd_uuid *cluuid,
2723                           struct obd_connect_data *data, void *localdata)
2724 {
2725         struct lvfs_run_ctxt saved;
2726         struct obd_export *exp;
2727         struct filter_export_data *fed;
2728         struct lsd_client_data *lcd = NULL;
2729         __u32 group;
2730         int rc;
2731         ENTRY;
2732
2733         if (conn == NULL || obd == NULL || cluuid == NULL)
2734                 RETURN(-EINVAL);
2735
2736         rc = class_connect(conn, obd, cluuid);
2737         if (rc)
2738                 RETURN(rc);
2739         exp = class_conn2export(conn);
2740         LASSERT(exp != NULL);
2741
2742         fed = &exp->exp_filter_data;
2743
2744         rc = filter_connect_internal(exp, data);
2745         if (rc)
2746                 GOTO(cleanup, rc);
2747
2748         filter_export_stats_init(obd, exp, localdata);
2749         if (obd->obd_replayable) {
2750                 OBD_ALLOC(lcd, sizeof(*lcd));
2751                 if (!lcd) {
2752                         CERROR("filter: out of memory for client data\n");
2753                         GOTO(cleanup, rc = -ENOMEM);
2754                 }
2755
2756                 memcpy(lcd->lcd_uuid, cluuid, sizeof(lcd->lcd_uuid));
2757                 fed->fed_lcd = lcd;
2758                 rc = filter_client_add(obd, exp, -1);
2759                 if (rc)
2760                         GOTO(cleanup, rc);
2761         }
2762
2763         group = data->ocd_group;
2764         if (group == 0)
2765                 GOTO(cleanup, rc);
2766
2767         CWARN("%s: Received MDS connection ("LPX64"); group %d\n",
2768               obd->obd_name, exp->exp_handle.h_cookie, group);
2769
2770         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2771         rc = filter_read_groups(obd, group, 1);
2772         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2773         if (rc != 0) {
2774                 CERROR("can't read group %u\n", group);
2775                 GOTO(cleanup, rc);
2776         }
2777
2778         GOTO(cleanup, rc);
2779
2780 cleanup:
2781         if (rc) {
2782                 if (lcd) {
2783                         OBD_FREE_PTR(lcd);
2784                         fed->fed_lcd = NULL;
2785                 }
2786                 class_disconnect(exp);
2787         } else {
2788                 class_export_put(exp);
2789         }
2790
2791         RETURN(rc);
2792 }
2793
2794 /* Do extra sanity checks for grant accounting.  We do this at connect,
2795  * disconnect, and statfs RPC time, so it shouldn't be too bad.  We can
2796  * always get rid of it or turn it off when we know accounting is good. */
2797 static void filter_grant_sanity_check(struct obd_device *obd, const char *func)
2798 {
2799         struct filter_export_data *fed;
2800         struct obd_export *exp;
2801         obd_size maxsize = obd->obd_osfs.os_blocks * obd->obd_osfs.os_bsize;
2802         obd_size tot_dirty = 0, tot_pending = 0, tot_granted = 0;
2803         obd_size fo_tot_dirty, fo_tot_pending, fo_tot_granted;
2804
2805         if (list_empty(&obd->obd_exports))
2806                 return;
2807
2808         /* We don't want to do this for large machines that do lots of
2809            mounts or unmounts.  It burns... */
2810         if (obd->obd_num_exports > 100)
2811                 return;
2812
2813         spin_lock(&obd->obd_osfs_lock);
2814         spin_lock(&obd->obd_dev_lock);
2815         list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
2816                 int error = 0;
2817                 fed = &exp->exp_filter_data;
2818                 if (fed->fed_grant < 0 || fed->fed_pending < 0 ||
2819                     fed->fed_dirty < 0)
2820                         error = 1;
2821                 if (maxsize > 0) { /* we may not have done a statfs yet */
2822                         LASSERTF(fed->fed_grant + fed->fed_pending <= maxsize,
2823                                  "%s: cli %s/%p %ld+%ld > "LPU64"\n", func,
2824                                  exp->exp_client_uuid.uuid, exp,
2825                                  fed->fed_grant, fed->fed_pending, maxsize);
2826                         LASSERTF(fed->fed_dirty <= maxsize,
2827                                  "%s: cli %s/%p %ld > "LPU64"\n", func,
2828                                  exp->exp_client_uuid.uuid, exp,
2829                                  fed->fed_dirty, maxsize);
2830                 }
2831                 if (error)
2832                         CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
2833                                obd->obd_name, exp->exp_client_uuid.uuid, exp,
2834                                fed->fed_dirty, fed->fed_pending,fed->fed_grant);
2835                 else
2836                         CDEBUG(D_CACHE, "%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
2837                                obd->obd_name, exp->exp_client_uuid.uuid, exp,
2838                                fed->fed_dirty, fed->fed_pending,fed->fed_grant);
2839                 tot_granted += fed->fed_grant + fed->fed_pending;
2840                 tot_pending += fed->fed_pending;
2841                 tot_dirty += fed->fed_dirty;
2842         }
2843         fo_tot_granted = obd->u.filter.fo_tot_granted;
2844         fo_tot_pending = obd->u.filter.fo_tot_pending;
2845         fo_tot_dirty = obd->u.filter.fo_tot_dirty;
2846         spin_unlock(&obd->obd_dev_lock);
2847         spin_unlock(&obd->obd_osfs_lock);
2848
2849         /* Do these assertions outside the spinlocks so we don't kill system */
2850         if (tot_granted != fo_tot_granted)
2851                 CERROR("%s: tot_granted "LPU64" != fo_tot_granted "LPU64"\n",
2852                        func, tot_granted, fo_tot_granted);
2853         if (tot_pending != fo_tot_pending)
2854                 CERROR("%s: tot_pending "LPU64" != fo_tot_pending "LPU64"\n",
2855                        func, tot_pending, fo_tot_pending);
2856         if (tot_dirty != fo_tot_dirty)
2857                 CERROR("%s: tot_dirty "LPU64" != fo_tot_dirty "LPU64"\n",
2858                        func, tot_dirty, fo_tot_dirty);
2859         if (tot_pending > tot_granted)
2860                 CERROR("%s: tot_pending "LPU64" > tot_granted "LPU64"\n",
2861                        func, tot_pending, tot_granted);
2862         if (tot_granted > maxsize)
2863                 CERROR("%s: tot_granted "LPU64" > maxsize "LPU64"\n",
2864                        func, tot_granted, maxsize);
2865         if (tot_dirty > maxsize)
2866                 CERROR("%s: tot_dirty "LPU64" > maxsize "LPU64"\n",
2867                        func, tot_dirty, maxsize);
2868 }
2869
2870 /* Remove this client from the grant accounting totals.  We also remove
2871  * the export from the obd device under the osfs and dev locks to ensure
2872  * that the filter_grant_sanity_check() calculations are always valid.
2873  * The client should do something similar when it invalidates its import. */
2874 static void filter_grant_discard(struct obd_export *exp)
2875 {
2876         struct obd_device *obd = exp->exp_obd;
2877         struct filter_obd *filter = &obd->u.filter;
2878         struct filter_export_data *fed = &exp->exp_filter_data;
2879
2880         spin_lock(&obd->obd_osfs_lock);
2881         spin_lock(&obd->obd_dev_lock);
2882         list_del_init(&exp->exp_obd_chain);
2883         spin_unlock(&obd->obd_dev_lock);
2884
2885         LASSERTF(filter->fo_tot_granted >= fed->fed_grant,
2886                  "%s: tot_granted "LPU64" cli %s/%p fed_grant %ld\n",
2887                  obd->obd_name, filter->fo_tot_granted,
2888                  exp->exp_client_uuid.uuid, exp, fed->fed_grant);
2889         filter->fo_tot_granted -= fed->fed_grant;
2890         LASSERTF(filter->fo_tot_pending >= fed->fed_pending,
2891                  "%s: tot_pending "LPU64" cli %s/%p fed_pending %ld\n",
2892                  obd->obd_name, filter->fo_tot_pending,
2893                  exp->exp_client_uuid.uuid, exp, fed->fed_pending);
2894         /* fo_tot_pending is handled in filter_grant_commit as bulk finishes */
2895         LASSERTF(filter->fo_tot_dirty >= fed->fed_dirty,
2896                  "%s: tot_dirty "LPU64" cli %s/%p fed_dirty %ld\n",
2897                  obd->obd_name, filter->fo_tot_dirty,
2898                  exp->exp_client_uuid.uuid, exp, fed->fed_dirty);
2899         filter->fo_tot_dirty -= fed->fed_dirty;
2900         fed->fed_dirty = 0;
2901         fed->fed_grant = 0;
2902
2903         spin_unlock(&obd->obd_osfs_lock);
2904 }
2905
2906 static int filter_destroy_export(struct obd_export *exp)
2907 {
2908         ENTRY;
2909
2910         if (exp->exp_filter_data.fed_pending)
2911                 CERROR("%s: cli %s/%p has %lu pending on destroyed export\n",
2912                        exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
2913                        exp, exp->exp_filter_data.fed_pending);
2914
2915         /* Not ported yet the b1_6 quota functionality
2916          * lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd);
2917          */
2918
2919         target_destroy_export(exp);
2920         ldlm_destroy_export(exp);
2921
2922         if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid))
2923                 RETURN(0);
2924
2925
2926         if (exp->exp_obd->obd_replayable)
2927                 filter_client_free(exp);
2928         else
2929                 fsfilt_sync(exp->exp_obd, exp->exp_obd->u.obt.obt_sb);
2930
2931         filter_grant_discard(exp);
2932         filter_fmd_cleanup(exp);
2933
2934         if (!(exp->exp_flags & OBD_OPT_FORCE))
2935                 filter_grant_sanity_check(exp->exp_obd, __FUNCTION__);
2936
2937         RETURN(0);
2938 }
2939
2940 static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp)
2941 {
2942         struct obd_llog_group *olg_min, *olg;
2943         struct filter_obd *filter;
2944         int worked = 0, group;
2945         struct llog_ctxt *ctxt;
2946         ENTRY;
2947
2948         filter = &obd->u.filter;
2949
2950         /* we can't sync log holding spinlock. also, we do not want to get
2951          * into livelock. so we do following: loop over MDS's exports in
2952          * group order and skip already synced llogs -bzzz */
2953         do {
2954                 /* look for group with min. number, but > worked */
2955                 olg_min = NULL;
2956                 group = 1 << 30;
2957                 spin_lock(&filter->fo_llog_list_lock);
2958                 list_for_each_entry(olg, &filter->fo_llog_list, olg_list) {
2959                         if (olg->olg_group <= worked) {
2960                                 /* this group is already synced */
2961                                 continue;
2962                         }
2963                         if (group < olg->olg_group) {
2964                                 /* we have group with smaller number to sync */
2965                                 continue;
2966                         }
2967                         /* store current minimal group */
2968                         olg_min = olg;
2969                         group = olg->olg_group;
2970                 }
2971                 spin_unlock(&filter->fo_llog_list_lock);
2972
2973                 if (olg_min == NULL)
2974                         break;
2975
2976                 worked = olg_min->olg_group;
2977                 if (olg_min->olg_exp &&
2978                     (dexp == olg_min->olg_exp || dexp == NULL)) {
2979                         int err;
2980                         ctxt = llog_group_get_ctxt(olg_min,
2981                                                 LLOG_MDS_OST_REPL_CTXT);
2982                         LASSERT(ctxt != NULL);
2983                         err = llog_sync(ctxt, olg_min->olg_exp);
2984                         llog_ctxt_put(ctxt);
2985                         if (err)
2986                                 CERROR("error flushing logs to MDS: rc %d\n",
2987                                        err);
2988                 }
2989         } while (olg_min != NULL);
2990 }
2991
2992 /* also incredibly similar to mds_disconnect */
2993 static int filter_disconnect(struct obd_export *exp)
2994 {
2995         struct obd_device *obd = exp->exp_obd;
2996         int rc;
2997         ENTRY;
2998
2999         LASSERT(exp);
3000         class_export_get(exp);
3001
3002         if (!(exp->exp_flags & OBD_OPT_FORCE))
3003                 filter_grant_sanity_check(obd, __FUNCTION__);
3004         filter_grant_discard(exp);
3005
3006         /* Disconnect early so that clients can't keep using export */
3007         rc = class_disconnect(exp);
3008         if (exp->exp_obd->obd_namespace != NULL)
3009                 ldlm_cancel_locks_for_export(exp);
3010
3011         fsfilt_sync(obd, obd->u.obt.obt_sb);
3012
3013         lprocfs_exp_cleanup(exp);
3014
3015         /* flush any remaining cancel messages out to the target */
3016         filter_sync_llogs(obd, exp);
3017         class_export_put(exp);
3018         RETURN(rc);
3019 }
3020
3021 /* reverse import is changed, sync all cancels */
3022 static void filter_revimp_update(struct obd_export *exp)
3023 {
3024         ENTRY;
3025
3026         LASSERT(exp);
3027         class_export_get(exp);
3028
3029         /* flush any remaining cancel messages out to the target */
3030         filter_sync_llogs(exp->exp_obd, exp);
3031         class_export_put(exp);
3032         EXIT;
3033 }
3034
3035 static int filter_ping(struct obd_export *exp)
3036 {
3037         filter_fmd_expire(exp);
3038
3039         return 0;
3040 }
3041
3042 struct dentry *__filter_oa2dentry(struct obd_device *obd, struct obdo *oa,
3043                                   const char *what, int quiet)
3044 {
3045         struct dentry *dchild = NULL;
3046         obd_gr group = 0;
3047
3048         if (oa->o_valid & OBD_MD_FLGROUP)
3049                 group = oa->o_gr;
3050
3051         dchild = filter_fid2dentry(obd, NULL, group, oa->o_id);
3052
3053         if (IS_ERR(dchild)) {
3054                 CERROR("%s error looking up object: "LPU64":"LPU64"\n",
3055                        what, group, oa->o_id);
3056                 RETURN(dchild);
3057         }
3058
3059         if (dchild->d_inode == NULL) {
3060                 if (!quiet)
3061                         CERROR("%s: %s on non-existent object: "LPU64"\n",
3062                                obd->obd_name, what, oa->o_id);
3063                 f_dput(dchild);
3064                 RETURN(ERR_PTR(-ENOENT));
3065         }
3066
3067         return dchild;
3068 }
3069
3070 static int filter_getattr(struct obd_export *exp, struct obd_info *oinfo)
3071 {
3072         struct dentry *dentry = NULL;
3073         struct obd_device *obd;
3074         int rc = 0;
3075         ENTRY;
3076
3077         rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
3078                               oinfo_capa(oinfo), CAPA_OPC_META_READ);
3079         if (rc)
3080                 RETURN(rc);
3081
3082         obd = class_exp2obd(exp);
3083         if (obd == NULL) {
3084                 CDEBUG(D_IOCTL, "invalid client export %p\n", exp);
3085                 RETURN(-EINVAL);
3086         }
3087
3088         dentry = filter_oa2dentry(obd, oinfo->oi_oa);
3089         if (IS_ERR(dentry))
3090                 RETURN(PTR_ERR(dentry));
3091
3092         /* Limit the valid bits in the return data to what we actually use */
3093         oinfo->oi_oa->o_valid = OBD_MD_FLID;
3094         obdo_from_inode(oinfo->oi_oa, dentry->d_inode, FILTER_VALID_FLAGS);
3095
3096         f_dput(dentry);
3097         RETURN(rc);
3098 }
3099
3100 /* this should be enabled/disabled in condition to enabled/disabled large
3101  * inodes (fast EAs) in backing store FS. */
3102 int filter_update_fidea(struct obd_export *exp, struct inode *inode,
3103                         void *handle, struct obdo *oa)
3104 {
3105         struct obd_device *obd = exp->exp_obd;
3106         int rc = 0;
3107         ENTRY;
3108
3109         if (oa->o_valid & OBD_MD_FLFID) {
3110                 struct filter_fid ff;
3111
3112                 if (!(oa->o_valid & OBD_MD_FLGROUP))
3113                         oa->o_gr = 0;
3114                 /* packing fid and converting it to LE for storing into EA.
3115                  * Here ->o_stripe_idx should be filled by LOV and rest of
3116                  * fields - by client. */
3117                 ff.ff_fid.id = cpu_to_le64(oa->o_fid);
3118                 ff.ff_fid.f_type = cpu_to_le32(oa->o_stripe_idx);
3119                 ff.ff_fid.generation = cpu_to_le32(oa->o_generation);
3120                 ff.ff_objid = cpu_to_le64(oa->o_id);
3121                 ff.ff_group = cpu_to_le64(oa->o_gr);
3122
3123                 CDEBUG(D_INODE, "storing filter fid EA ("LPU64"/%u/%u"
3124                        LPU64"/"LPU64")\n", oa->o_fid, oa->o_stripe_idx,
3125                        oa->o_generation, oa->o_id, oa->o_gr);
3126
3127                 rc = fsfilt_set_md(obd, inode, handle, &ff, sizeof(ff), "fid");
3128                 if (rc)
3129                         CERROR("store fid in object failed! rc: %d\n", rc);
3130         } else {
3131                 CDEBUG(D_HA, "OSS object without fid info!\n");
3132         }
3133
3134         RETURN(rc);
3135 }
3136
3137 /* this is called from filter_truncate() until we have filter_punch() */
3138 int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry,
3139                             struct obdo *oa, struct obd_trans_info *oti)
3140 {
3141         unsigned int orig_ids[MAXQUOTAS] = {0, 0};
3142         struct llog_cookie *fcc = NULL;
3143         struct filter_obd *filter;
3144         int rc, err, locked = 0, sync = 0;
3145         loff_t old_size = 0;
3146         unsigned int ia_valid;
3147         struct inode *inode;
3148         struct iattr iattr;
3149         void *handle;
3150         ENTRY;
3151
3152         LASSERT(dentry != NULL);
3153         LASSERT(!IS_ERR(dentry));
3154
3155         inode = dentry->d_inode;
3156         LASSERT(inode != NULL);
3157
3158         filter = &exp->exp_obd->u.filter;
3159         iattr_from_obdo(&iattr, oa, oa->o_valid);
3160         ia_valid = iattr.ia_valid;
3161
3162         if (oa->o_valid & OBD_MD_FLCOOKIE) {
3163                 OBD_ALLOC(fcc, sizeof(*fcc));
3164                 if (fcc != NULL)
3165                         *fcc = oa->o_lcookie;
3166         }
3167
3168         if (ia_valid & ATTR_SIZE || ia_valid & (ATTR_UID | ATTR_GID)) {
3169                 old_size = i_size_read(inode);
3170                 DQUOT_INIT(inode);
3171                 LOCK_INODE_MUTEX(inode);
3172                 locked = 1;
3173         }
3174
3175         /* If the inode still has SUID+SGID bits set (see filter_precreate())
3176          * then we will accept the UID+GID sent by the client during write for
3177          * initializing the ownership of this inode.  We only allow this to
3178          * happen once so clear these bits in setattr. In 2.6 kernels it is
3179          * possible to get ATTR_UID and ATTR_GID separately, so we only clear
3180          * the flags that are actually being set. */
3181         if (ia_valid & (ATTR_UID | ATTR_GID)) {
3182                 CDEBUG(D_INODE, "update UID/GID to %lu/%lu\n",
3183                        (unsigned long)oa->o_uid, (unsigned long)oa->o_gid);
3184
3185                 if ((inode->i_mode & S_ISUID) && (ia_valid & ATTR_UID)) {
3186                         if (!(ia_valid & ATTR_MODE)) {
3187                                 iattr.ia_mode = inode->i_mode;
3188                                 iattr.ia_valid |= ATTR_MODE;
3189                         }
3190                         iattr.ia_mode &= ~S_ISUID;
3191                 }
3192                 if ((inode->i_mode & S_ISGID) && (ia_valid & ATTR_GID)) {
3193                         if (!(iattr.ia_valid & ATTR_MODE)) {
3194                                 iattr.ia_mode = inode->i_mode;
3195                                 iattr.ia_valid |= ATTR_MODE;
3196                         }
3197                         iattr.ia_mode &= ~S_ISGID;
3198                 }
3199
3200                 orig_ids[USRQUOTA] = inode->i_uid;
3201                 orig_ids[GRPQUOTA] = inode->i_gid;
3202                 handle = fsfilt_start_log(exp->exp_obd, inode,
3203                                           FSFILT_OP_SETATTR, oti, 1);
3204                 if (IS_ERR(handle))
3205                         GOTO(out_unlock, rc = PTR_ERR(handle));
3206
3207                 /* update inode EA only once when inode is suid bit marked. As
3208                  * on 2.6.x UID and GID may be set separately, we check here
3209                  * only one of them to avoid double setting. */
3210                 if (inode->i_mode & S_ISUID)
3211                         filter_update_fidea(exp, inode, handle, oa);
3212         } else {
3213                 handle = fsfilt_start(exp->exp_obd, inode,
3214                                       FSFILT_OP_SETATTR, oti);
3215                 if (IS_ERR(handle))
3216                         GOTO(out_unlock, rc = PTR_ERR(handle));
3217         }
3218         if (oa->o_valid & OBD_MD_FLFLAGS) {
3219                 rc = fsfilt_iocontrol(exp->exp_obd, inode, NULL,
3220                                       EXT3_IOC_SETFLAGS, (long)&oa->o_flags);
3221         } else {
3222                 rc = fsfilt_setattr(exp->exp_obd, dentry, handle, &iattr, 1);
3223                 if (fcc != NULL)
3224                         /* set cancel cookie callback function */
3225                         sync = fsfilt_add_journal_cb(exp->exp_obd, 0, handle,
3226                                                      filter_cancel_cookies_cb,
3227                                                      fcc);
3228         }
3229
3230         if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_CREDITS))
3231                 fsfilt_extend(exp->exp_obd, inode, 0, handle);
3232
3233         /* The truncate might have used up our transaction credits.  Make
3234          * sure we have one left for the last_rcvd update. */
3235         err = fsfilt_extend(exp->exp_obd, inode, 1, handle);
3236
3237         rc = filter_finish_transno(exp, oti, rc, sync);
3238         if (sync) {
3239                 filter_cancel_cookies_cb(exp->exp_obd, 0, fcc, rc);
3240                 fcc = NULL;
3241         }
3242
3243         err = fsfilt_commit(exp->exp_obd, inode, handle, 0);
3244         if (err) {
3245                 CERROR("error on commit, err = %d\n", err);
3246                 if (!rc)
3247                         rc = err;
3248         } else {
3249                 fcc = NULL;
3250         }
3251
3252         /* For a partial-page truncate flush the page to disk immediately
3253          * to avoid data corruption during direct disk write. b=17397 */
3254         if (!sync && (iattr.ia_valid & ATTR_SIZE) &&
3255             old_size != iattr.ia_size && (iattr.ia_size & ~CFS_PAGE_MASK)) {
3256                 err = filemap_fdatawrite_range(inode->i_mapping, iattr.ia_size,
3257                                                iattr.ia_size + 1);
3258                 if (!rc)
3259                         rc = err;
3260         }
3261
3262         if (locked) {
3263                 UNLOCK_INODE_MUTEX(inode);
3264                 locked = 0;
3265         }
3266
3267         EXIT;
3268 out_unlock:
3269         if (locked)
3270                 UNLOCK_INODE_MUTEX(inode);
3271
3272         if (fcc)
3273                 OBD_FREE(fcc, sizeof(*fcc));
3274
3275         /* trigger quota release */
3276         if (ia_valid & (ATTR_SIZE | ATTR_UID | ATTR_GID)) {
3277                 unsigned int cur_ids[MAXQUOTAS] = {oa->o_uid, oa->o_gid};
3278                 int rc2 = lquota_adjust(filter_quota_interface_ref,
3279                                         exp->exp_obd, cur_ids,
3280                                         orig_ids, rc, FSFILT_OP_SETATTR);
3281                 CDEBUG(rc2 ? D_ERROR : D_QUOTA,
3282                        "filter adjust qunit. (rc:%d)\n", rc2);
3283         }
3284         return rc;
3285 }
3286
3287 /* this is called from filter_truncate() until we have filter_punch() */
3288 int filter_setattr(struct obd_export *exp, struct obd_info *oinfo,
3289                    struct obd_trans_info *oti)
3290 {
3291         struct ldlm_res_id res_id;
3292         struct filter_mod_data *fmd;
3293         struct lvfs_run_ctxt saved;
3294         struct filter_obd *filter;
3295         struct ldlm_resource *res;
3296         struct dentry *dentry;
3297         int rc;
3298         ENTRY;
3299
3300         osc_build_res_name(oinfo->oi_oa->o_id, oinfo->oi_oa->o_gr, &res_id);
3301         rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
3302                               oinfo_capa(oinfo), CAPA_OPC_META_WRITE);
3303         if (rc)
3304                 RETURN(rc);
3305
3306         /* This would be very bad - accidentally truncating a file when
3307          * changing the time or similar - bug 12203. */
3308         if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE &&
3309             oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) {
3310                 static char mdsinum[48];
3311
3312                 if (oinfo->oi_oa->o_valid & OBD_MD_FLFID)
3313                         snprintf(mdsinum, sizeof(mdsinum) - 1,
3314                                  " of inode "LPU64"/%u", oinfo->oi_oa->o_fid,
3315                                  oinfo->oi_oa->o_generation);
3316                 else
3317                         mdsinum[0] = '\0';
3318
3319                 CERROR("%s: setattr from %s trying to truncate objid "LPU64
3320                        " %s\n",
3321                        exp->exp_obd->obd_name, obd_export_nid2str(exp),
3322                        oinfo->oi_oa->o_id, mdsinum);
3323                 RETURN(-EPERM);
3324         }
3325
3326         dentry = __filter_oa2dentry(exp->exp_obd, oinfo->oi_oa,
3327                                     __FUNCTION__, 1);
3328         if (IS_ERR(dentry))
3329                 RETURN(PTR_ERR(dentry));
3330
3331         filter = &exp->exp_obd->u.filter;
3332         push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
3333         lock_kernel();
3334
3335         if (oinfo->oi_oa->o_valid &
3336             (OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME)) {
3337                 fmd = filter_fmd_get(exp,oinfo->oi_oa->o_id,oinfo->oi_oa->o_gr);
3338                 if (fmd && fmd->fmd_mactime_xid < oti->oti_xid)
3339                         fmd->fmd_mactime_xid = oti->oti_xid;
3340                 filter_fmd_put(exp, fmd);
3341         }
3342
3343         /* setting objects attributes (including owner/group) */
3344         rc = filter_setattr_internal(exp, dentry, oinfo->oi_oa, oti);
3345         if (rc)
3346                 GOTO(out_unlock, rc);
3347
3348         res = ldlm_resource_get(exp->exp_obd->obd_namespace, NULL,
3349                                 &res_id, LDLM_EXTENT, 0);
3350
3351         if (res != NULL) {
3352                 LDLM_RESOURCE_ADDREF(res);
3353                 rc = ldlm_res_lvbo_update(res, NULL, 0, 0);
3354                 LDLM_RESOURCE_DELREF(res);
3355                 ldlm_resource_putref(res);
3356         }
3357
3358         oinfo->oi_oa->o_valid = OBD_MD_FLID;
3359
3360         /* Quota release need uid/gid info */
3361         obdo_from_inode(oinfo->oi_oa, dentry->d_inode,
3362                         FILTER_VALID_FLAGS | OBD_MD_FLUID | OBD_MD_FLGID);
3363
3364         EXIT;
3365 out_unlock:
3366         unlock_kernel();
3367         f_dput(dentry);
3368         pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
3369         return rc;
3370 }
3371
3372 /* XXX identical to osc_unpackmd */
3373 static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
3374                            struct lov_mds_md *lmm, int lmm_bytes)
3375 {
3376         int lsm_size;
3377         ENTRY;
3378
3379         if (lmm != NULL) {
3380                 if (lmm_bytes < sizeof (*lmm)) {
3381                         CERROR("lov_mds_md too small: %d, need %d\n",
3382                                lmm_bytes, (int)sizeof(*lmm));
3383                         RETURN(-EINVAL);
3384                 }
3385                 /* XXX LOV_MAGIC etc check? */
3386
3387                 if (lmm->lmm_object_id == cpu_to_le64(0)) {
3388                         CERROR("lov_mds_md: zero lmm_object_id\n");
3389                         RETURN(-EINVAL);
3390                 }
3391         }
3392
3393         lsm_size = lov_stripe_md_size(1);
3394         if (lsmp == NULL)
3395                 RETURN(lsm_size);
3396
3397         if (*lsmp != NULL && lmm == NULL) {
3398                 OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
3399                 OBD_FREE(*lsmp, lsm_size);
3400                 *lsmp = NULL;
3401                 RETURN(0);
3402         }
3403
3404         if (*lsmp == NULL) {
3405                 OBD_ALLOC(*lsmp, lsm_size);
3406                 if (*lsmp == NULL)
3407                         RETURN(-ENOMEM);
3408
3409                 OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
3410                 if ((*lsmp)->lsm_oinfo[0] == NULL) {
3411                         OBD_FREE(*lsmp, lsm_size);
3412                         RETURN(-ENOMEM);
3413                 }
3414                 loi_init((*lsmp)->lsm_oinfo[0]);
3415         }
3416
3417         if (lmm != NULL) {
3418                 /* XXX zero *lsmp? */
3419                 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
3420                 LASSERT((*lsmp)->lsm_object_id);
3421         }
3422
3423         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
3424
3425         RETURN(lsm_size);
3426 }
3427
3428 /* caller must hold fo_create_locks[oa->o_gr] */
3429 static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
3430                                       struct filter_obd *filter)
3431 {
3432         struct obdo doa; /* XXX obdo on stack */
3433         obd_id last, id;
3434         int rc;
3435         ENTRY;
3436
3437         LASSERT(oa);
3438         LASSERT(oa->o_gr != 0);
3439         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
3440         LASSERT(down_trylock(&filter->fo_create_locks[oa->o_gr]) != 0);
3441
3442         memset(&doa, 0, sizeof(doa));
3443
3444         doa.o_valid |= OBD_MD_FLGROUP;
3445         doa.o_gr = oa->o_gr;
3446         doa.o_mode = S_IFREG;
3447
3448         if (!test_bit(doa.o_gr, &filter->fo_destroys_in_progress)) {
3449                 CERROR("%s:["LPU64"] destroys_in_progress already cleared\n",
3450                        exp->exp_obd->obd_name, doa.o_gr);
3451                 RETURN(0);
3452         }
3453
3454         last = filter_last_id(filter, doa.o_gr);
3455
3456         CWARN("%s: deleting orphan objects from "LPU64" to "LPU64"\n",
3457                exp->exp_obd->obd_name, oa->o_id + 1, last);
3458
3459         for (id = last; id > oa->o_id; id--) {
3460                 doa.o_id = id;
3461                 rc = filter_destroy(exp, &doa, NULL, NULL, NULL);
3462                 if (rc && rc != -ENOENT) /* this is pretty fatal... */
3463                         CEMERG("error destroying precreate objid "LPU64": %d\n",
3464                                id, rc);
3465                 filter_set_last_id(filter, id - 1, doa.o_gr);
3466                 /* update last_id on disk periodically so that if we restart
3467                  * we don't need to re-scan all of the just-deleted objects. */
3468                 if ((id & 511) == 0)
3469                         filter_update_last_objid(exp->exp_obd, doa.o_gr, 0);
3470         }
3471
3472         CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n",
3473                exp->exp_obd->obd_name, doa.o_gr, oa->o_id);
3474
3475         rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
3476         clear_bit(doa.o_gr, &filter->fo_destroys_in_progress);
3477
3478         RETURN(rc);
3479 }
3480
3481 static int filter_precreate(struct obd_device *obd, struct obdo *oa,
3482                             obd_gr group, int *num);
3483 /* returns a negative error or a nonnegative number of files to create */
3484 static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa,
3485                                    obd_gr group, struct obd_trans_info *oti)
3486 {
3487         struct obd_device *obd = exp->exp_obd;
3488         struct filter_obd *filter = &obd->u.filter;
3489         int diff, rc;
3490         ENTRY;
3491
3492         /* delete orphans request */
3493         if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_DELORPHAN)){
3494                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
3495                         CERROR("%s: dropping old orphan cleanup request\n",
3496                                obd->obd_name);
3497                         RETURN(0);
3498                 }
3499                 /* This causes inflight precreates to abort and drop lock */
3500                 set_bit(group, &filter->fo_destroys_in_progress);
3501                 down(&filter->fo_create_locks[group]);
3502                 if (!test_bit(group, &filter->fo_destroys_in_progress)) {
3503                         CERROR("%s:["LPU64"] destroys_in_progress already cleared\n",
3504                                exp->exp_obd->obd_name, group);
3505                         up(&filter->fo_create_locks[group]);
3506                         RETURN(0);
3507                 }
3508                 diff = oa->o_id - filter_last_id(filter, group);
3509                 CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n",
3510                        filter_last_id(filter, group), diff);
3511
3512                 if (-diff > OST_MAX_PRECREATE) {
3513                         CERROR("%s: ignoring bogus orphan destroy request: "
3514                                "obdid "LPU64" last_id "LPU64"\n", obd->obd_name,
3515                                oa->o_id, filter_last_id(filter, group));
3516                         /* FIXME: should reset precreate_next_id on MDS */
3517                         GOTO(out, rc = -EINVAL);
3518                 }
3519                 if (diff < 0) {
3520                         rc = filter_destroy_precreated(exp, oa, filter);
3521                         if (rc)
3522                                 CERROR("%s: unable to write lastobjid, but "
3523                                        "orphans were deleted\n", obd->obd_name);
3524                         GOTO(out, rc);
3525                 } else {
3526                         /* XXX: Used by MDS for the first time! */
3527                         clear_bit(group, &filter->fo_destroys_in_progress);
3528                 }
3529         } else {
3530                 down(&filter->fo_create_locks[group]);
3531                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
3532                         CERROR("%s: dropping old precreate request\n",
3533                                obd->obd_name);
3534                         GOTO(out, rc = 0);
3535                 }
3536                 /* only precreate if group == 0 and o_id is specified */
3537                 if (group < FILTER_GROUP_MDS0 || oa->o_id == 0)
3538                         diff = 1;
3539                 else
3540                         diff = oa->o_id - filter_last_id(filter, group);
3541                 CDEBUG(D_RPCTRACE, "filter_last_id() = "LPU64" -> diff = %d\n",
3542                        filter_last_id(filter, group), diff);
3543
3544                 LASSERTF(diff >= 0,"%s: "LPU64" - "LPU64" = %d\n",obd->obd_name,
3545                          oa->o_id, filter_last_id(filter, group), diff);
3546         }
3547
3548         if (diff > 0) {
3549                 oa->o_id = filter_last_id(&obd->u.filter, group);
3550                 rc = filter_precreate(obd, oa, group, &diff);
3551                 oa->o_id = filter_last_id(&obd->u.filter, group);
3552                 oa->o_gr = group;
3553                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
3554                 GOTO(out, rc);
3555         }
3556         /* else diff == 0 */
3557         GOTO(out, rc = 0);
3558 out:
3559         up(&filter->fo_create_locks[group]);
3560         return rc;
3561 }
3562
3563 static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3564                          __u64 max_age, __u32 flags)
3565 {
3566         struct filter_obd *filter = &obd->u.filter;
3567         int blockbits = obd->u.obt.obt_sb->s_blocksize_bits;
3568         int rc;
3569         ENTRY;
3570
3571         /* at least try to account for cached pages.  its still racey and
3572          * might be under-reporting if clients haven't announced their
3573          * caches with brw recently */
3574         spin_lock(&obd->obd_osfs_lock);
3575         rc = fsfilt_statfs(obd, obd->u.obt.obt_sb, max_age);
3576         memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
3577         spin_unlock(&obd->obd_osfs_lock);
3578
3579         CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64
3580                " pending "LPU64" free "LPU64" avail "LPU64"\n",
3581                filter->fo_tot_dirty, filter->fo_tot_granted,
3582                filter->fo_tot_pending,
3583                osfs->os_bfree << blockbits, osfs->os_bavail << blockbits);
3584
3585         filter_grant_sanity_check(obd, __FUNCTION__);
3586
3587         osfs->os_bavail -= min(osfs->os_bavail, GRANT_FOR_LLOG(obd) +
3588                                ((filter->fo_tot_dirty + filter->fo_tot_pending +
3589                                  osfs->os_bsize - 1) >> blockbits));
3590
3591         /* set EROFS to state field if FS is mounted as RDONLY. The goal is to
3592          * stop creating files on MDS if OST is not good shape to create
3593          * objects.*/
3594         osfs->os_state = (filter->fo_obt.obt_sb->s_flags & MS_RDONLY) ?
3595                 EROFS : 0;
3596         RETURN(rc);
3597 }
3598
3599 static int filter_use_existing_obj(struct obd_device *obd,
3600                                    struct dentry *dchild, void **handle,
3601                                    int *cleanup_phase)
3602 {
3603         struct inode *inode = dchild->d_inode;
3604         struct iattr iattr;
3605         int rc;
3606
3607         if ((inode->i_mode & (S_ISUID | S_ISGID)) == (S_ISUID|S_ISGID))
3608                 return 0;
3609
3610         *handle = fsfilt_start_log(obd, inode, FSFILT_OP_SETATTR, NULL, 1);
3611         if (IS_ERR(*handle))
3612                 return PTR_ERR(*handle);
3613
3614         iattr.ia_valid = ATTR_MODE;
3615         iattr.ia_mode = S_ISUID | S_ISGID |0666;
3616         rc = fsfilt_setattr(obd, dchild, *handle, &iattr, 1);
3617         if (rc == 0)
3618                 *cleanup_phase = 3;
3619
3620         return rc;
3621 }
3622
3623
3624 /* We rely on the fact that only one thread will be creating files in a given
3625  * group at a time, which is why we don't need an atomic filter_get_new_id.
3626  * Even if we had that atomic function, the following race would exist:
3627  *
3628  * thread 1: gets id x from filter_next_id
3629  * thread 2: gets id (x + 1) from filter_next_id
3630  * thread 2: creates object (x + 1)
3631  * thread 1: tries to create object x, gets -ENOSPC
3632  *
3633  * Caller must hold fo_create_locks[group]
3634  */
3635 static int filter_precreate(struct obd_device *obd, struct obdo *oa,
3636                             obd_gr group, int *num)
3637 {
3638         struct dentry *dchild = NULL, *dparent = NULL;
3639         struct filter_obd *filter;
3640         struct obd_statfs *osfs;
3641         int err = 0, rc = 0, recreate_obj = 0, i;
3642         cfs_time_t enough_time = cfs_time_shift(DISK_TIMEOUT/2);
3643         obd_id next_id;
3644         void *handle = NULL;
3645         ENTRY;
3646
3647         filter = &obd->u.filter;
3648
3649         LASSERT(down_trylock(&filter->fo_create_locks[group]) != 0);
3650
3651         OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_PRECREATE, obd_timeout / 2);
3652
3653         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
3654             (oa->o_flags & OBD_FL_RECREATE_OBJS)) {
3655                 recreate_obj = 1;
3656         } else {
3657                 OBD_ALLOC(osfs, sizeof(*osfs));
3658                 if (osfs == NULL)
3659                         RETURN(-ENOMEM);
3660                 rc = filter_statfs(obd, osfs, cfs_time_current_64() - HZ, 0);
3661                 if (rc == 0 && osfs->os_bavail < (osfs->os_blocks >> 10)) {
3662                         CDEBUG(D_RPCTRACE,"%s: not enough space for create "
3663                                LPU64"\n", obd->obd_name, osfs->os_bavail <<
3664                                filter->fo_vfsmnt->mnt_sb->s_blocksize_bits);
3665                         *num = 0;
3666                         rc = -ENOSPC;
3667                 }
3668                 OBD_FREE(osfs, sizeof(*osfs));
3669                 if (rc)
3670                         RETURN(rc);
3671         }
3672
3673         CDEBUG(D_RPCTRACE, "%s: precreating %d objects in group "LPU64
3674                " at "LPU64"\n", obd->obd_name, *num, group, oa->o_id);
3675
3676         for (i = 0; i < *num && err == 0; i++) {
3677                 int cleanup_phase = 0;
3678
3679                 if (test_bit(group, &filter->fo_destroys_in_progress)) {
3680                         CWARN("%s: create aborted by destroy\n",
3681                               obd->obd_name);
3682    &