lustre/obdfilter/filter.c

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