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