lustre/obdfilter/filter.c

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