lustre/obdfilter/filter.c

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