lustre/obdfilter/filter.c

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