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