lustre/obdfilter/filter.c

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