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