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