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_sem 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_sem to avoid single
  33  *            threaded operation on the OST.
  34  */
  35
  36 #define DEBUG_SUBSYSTEM S_FILTER
  37
  38 #include <linux/config.h>
  39 #include <linux/module.h>
  40 #include <linux/fs.h>
  41 #include <linux/dcache.h>
  42 #include <linux/init.h>
  43 #include <linux/version.h>
  44 #include <linux/jbd.h>
  45 #include <linux/ext3_fs.h>
  46 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
  47 # include <linux/mount.h>
  48 # include <linux/buffer_head.h>
  49 # include <linux/bio.h>
  50 #endif
  51
  52 #include <linux/obd_class.h>
  53 #include <linux/obd_lov.h>
  54 #include <linux/obd_ost.h>
  55 #include <linux/lustre_dlm.h>
  56 #include <linux/lustre_fsfilt.h>
  57 #include <linux/lprocfs_status.h>
  58 #include <linux/lustre_log.h>
  59 #include <linux/lustre_commit_confd.h>
  60 #include <libcfs/list.h>
  61
  62 #include <linux/lustre_smfs.h>
  63 #include "filter_internal.h"
  64
  65 /* Group 0 is no longer a legal group, to catch uninitialized IDs */
  66 #define FILTER_MIN_GROUPS 3
  67
  68 static struct lvfs_callback_ops filter_lvfs_ops;
  69
  70 static int filter_destroy(struct obd_export *exp, struct obdo *oa,
  71                           struct lov_stripe_md *ea, struct obd_trans_info *);
  72 struct obd_llogs *filter_grab_llog_for_group(struct obd_device *,
  73                                              int, struct obd_export *);
  74
  75 static void filter_commit_cb(struct obd_device *obd, __u64 transno,
  76                              void *cb_data, int error)
  77 {
  78         obd_transno_commit_cb(obd, transno, error);
  79 }
  80
  81 /* Assumes caller has already pushed us into the kernel context. */
  82 int filter_finish_transno(struct obd_export *exp, struct obd_trans_info *oti,
  83                           int rc)
  84 {
  85         struct filter_obd *filter = &exp->exp_obd->u.filter;
  86         struct filter_export_data *fed = &exp->exp_filter_data;
  87         struct filter_client_data *fcd = fed->fed_fcd;
  88         __u64 last_rcvd;
  89         loff_t off;
  90         int err, log_pri = D_HA;
  91
  92         /* Propagate error code. */
  93         if (rc)
  94                 RETURN(rc);
  95
  96         if (!exp->exp_obd->obd_replayable || oti == NULL)
  97                 RETURN(rc);
  98
  99         /* we don't allocate new transnos for replayed requests */
 100         if (oti->oti_transno == 0) {
 101                 spin_lock(&filter->fo_translock);
 102                 last_rcvd = le64_to_cpu(filter->fo_fsd->fsd_last_transno) + 1;
 103                 filter->fo_fsd->fsd_last_transno = cpu_to_le64(last_rcvd);
 104                 spin_unlock(&filter->fo_translock);
 105                 oti->oti_transno = last_rcvd;
 106         } else {
 107                 spin_lock(&filter->fo_translock);
 108                 last_rcvd = oti->oti_transno;
 109                 if (last_rcvd > le64_to_cpu(filter->fo_fsd->fsd_last_transno))
 110                         filter->fo_fsd->fsd_last_transno =
 111                                 cpu_to_le64(last_rcvd);
 112                 spin_unlock(&filter->fo_translock);
 113         }
 114
 115         fcd->fcd_last_rcvd = cpu_to_le64(last_rcvd);
 116
 117         /* could get xid from oti, if it's ever needed */
 118         fcd->fcd_last_xid = 0;
 119
 120         off = fed->fed_lr_off;
 121
 122         fsfilt_add_journal_cb(exp->exp_obd, filter->fo_sb, last_rcvd,
 123                               oti->oti_handle, filter_commit_cb, NULL);
 124
 125         err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp, fcd,
 126                                   sizeof(*fcd), &off, 0);
 127         if (err) {
 128                 log_pri = D_ERROR;
 129                 if (rc == 0)
 130                         rc = err;
 131         }
 132
 133         CDEBUG(log_pri, "wrote trans "LPU64" for client %s at #%d: err = %d\n",
 134                last_rcvd, fcd->fcd_uuid, fed->fed_lr_idx, err);
 135
 136         RETURN(rc);
 137 }
 138
 139 void f_dput(struct dentry *dentry)
 140 {
 141         /* Can't go inside filter_ddelete because it can block */
 142         CDEBUG(D_INODE, "putting %s: %p, count = %d\n",
 143                dentry->d_name.name, dentry, atomic_read(&dentry->d_count) - 1);
 144         LASSERT(atomic_read(&dentry->d_count) > 0);
 145
 146         dput(dentry);
 147 }
 148
 149 /* Add client data to the FILTER.  We use a bitmap to locate a free space
 150  * in the last_rcvd file if cl_idx is -1 (i.e. a new client).
 151  * Otherwise, we have just read the data from the last_rcvd file and
 152  * we know its offset. */
 153 static int filter_client_add(struct obd_device *obd, struct filter_obd *filter,
 154                              struct filter_export_data *fed, int cl_idx)
 155 {
 156         unsigned long *bitmap = filter->fo_last_rcvd_slots;
 157         int new_client = (cl_idx == -1);
 158         ENTRY;
 159
 160         LASSERT(bitmap != NULL);
 161
 162         /* XXX if fcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
 163         if (!strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid))
 164                 RETURN(0);
 165
 166         /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
 167          * there's no need for extra complication here
 168          */
 169         if (new_client) {
 170                 cl_idx = find_first_zero_bit(bitmap, FILTER_LR_MAX_CLIENTS);
 171         repeat:
 172                 if (cl_idx >= FILTER_LR_MAX_CLIENTS) {
 173                         CERROR("no client slots - fix FILTER_LR_MAX_CLIENTS\n");
 174                         RETURN(-ENOMEM);
 175                 }
 176                 if (test_and_set_bit(cl_idx, bitmap)) {
 177                         CERROR("FILTER client %d: found bit is set in bitmap\n",
 178                                cl_idx);
 179                         cl_idx = find_next_zero_bit(bitmap,
 180                                                     FILTER_LR_MAX_CLIENTS,
 181                                                     cl_idx);
 182                         goto repeat;
 183                 }
 184         } else {
 185                 if (test_and_set_bit(cl_idx, bitmap)) {
 186                         CERROR("FILTER client %d: bit already set in bitmap!\n",
 187                                cl_idx);
 188                         LBUG();
 189                 }
 190         }
 191
 192         fed->fed_lr_idx = cl_idx;
 193         fed->fed_lr_off = le32_to_cpu(filter->fo_fsd->fsd_client_start) +
 194                 cl_idx * le16_to_cpu(filter->fo_fsd->fsd_client_size);
 195
 196         CDEBUG(D_INFO, "client at index %d (%llu) with UUID '%s' added\n",
 197                fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid);
 198
 199         if (new_client) {
 200                 struct lvfs_run_ctxt saved;
 201                 loff_t off = fed->fed_lr_off;
 202                 int err;
 203                 void *handle;
 204
 205                 CDEBUG(D_INFO, "writing client fcd at idx %u (%llu) (len %u)\n",
 206                        fed->fed_lr_idx,off,(unsigned int)sizeof(*fed->fed_fcd));
 207
 208                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 209                 /* Transaction needed to fix bug 1403 */
 210                 handle = fsfilt_start(obd,
 211                                       filter->fo_rcvd_filp->f_dentry->d_inode,
 212                                       FSFILT_OP_SETATTR, NULL);
 213                 if (IS_ERR(handle)) {
 214                         err = PTR_ERR(handle);
 215                         CERROR("unable to start transaction: rc %d\n", err);
 216                 } else {
 217                         err = fsfilt_write_record(obd, filter->fo_rcvd_filp,
 218                                                   fed->fed_fcd,
 219                                                   sizeof(*fed->fed_fcd),
 220                                                   &off, 1);
 221                         fsfilt_commit(obd, filter->fo_sb,
 222                                       filter->fo_rcvd_filp->f_dentry->d_inode,
 223                                       handle, 1);
 224                 }
 225                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 226
 227                 if (err) {
 228                         CERROR("error writing %s client idx %u: rc %d\n",
 229                                LAST_RCVD, fed->fed_lr_idx, err);
 230                         RETURN(err);
 231                 }
 232         }
 233         RETURN(0);
 234 }
 235
 236 static int filter_client_free(struct obd_export *exp, int flags)
 237 {
 238         struct filter_export_data *fed = &exp->exp_filter_data;
 239         struct filter_obd *filter = &exp->exp_obd->u.filter;
 240         struct obd_device *obd = exp->exp_obd;
 241         struct filter_client_data zero_fcd;
 242         struct lvfs_run_ctxt saved;
 243         int rc;
 244         loff_t off;
 245         ENTRY;
 246
 247         if (fed->fed_fcd == NULL)
 248                 RETURN(0);
 249
 250         if (flags & OBD_OPT_FAILOVER)
 251                 GOTO(free, 0);
 252
 253         /* XXX if fcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
 254         if (strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid ) == 0)
 255                 GOTO(free, 0);
 256
 257         LASSERT(filter->fo_last_rcvd_slots != NULL);
 258
 259         off = fed->fed_lr_off;
 260
 261         CDEBUG(D_INFO, "freeing client at idx %u (%lld) with UUID '%s'\n",
 262                fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid);
 263
 264         /* Clear the bit _after_ zeroing out the client so we don't
 265            race with filter_client_add and zero out new clients.*/
 266         if (!test_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
 267                 CERROR("FILTER client %u: bit already clear in bitmap!!\n",
 268                        fed->fed_lr_idx);
 269                 LBUG();
 270         }
 271
 272         memset(&zero_fcd, 0, sizeof zero_fcd);
 273         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 274         rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, &zero_fcd,
 275                                  sizeof(zero_fcd), &off, 1);
 276         if (rc == 0)
 277                 /* update server's transno */
 278                 filter_update_server_data(obd, filter->fo_rcvd_filp,
 279                                           filter->fo_fsd, 1);
 280         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 281
 282         CDEBUG(rc == 0 ? D_INFO : D_ERROR,
 283                "zeroing disconnecting client %s at idx %u (%llu) in %s rc %d\n",
 284                fed->fed_fcd->fcd_uuid, fed->fed_lr_idx, fed->fed_lr_off,
 285                LAST_RCVD, rc);
 286
 287         if (!test_and_clear_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
 288                 CERROR("FILTER client %u: bit already clear in bitmap!!\n",
 289                        fed->fed_lr_idx);
 290                 LBUG();
 291         }
 292
 293 free:
 294         OBD_FREE(fed->fed_fcd, sizeof(*fed->fed_fcd));
 295
 296         RETURN(0);
 297 }
 298
 299 static int filter_free_server_data(struct filter_obd *filter)
 300 {
 301         OBD_FREE(filter->fo_fsd, sizeof(*filter->fo_fsd));
 302         filter->fo_fsd = NULL;
 303         OBD_FREE(filter->fo_last_rcvd_slots, FILTER_LR_MAX_CLIENTS/8);
 304         filter->fo_last_rcvd_slots = NULL;
 305         return 0;
 306 }
 307
 308 /* assumes caller is already in kernel ctxt */
 309 int filter_update_server_data(struct obd_device *obd, struct file *filp,
 310                               struct filter_server_data *fsd, int force_sync)
 311 {
 312         loff_t off = 0;
 313         int rc;
 314         ENTRY;
 315
 316         CDEBUG(D_INODE, "server uuid      : %s\n", fsd->fsd_uuid);
 317         CDEBUG(D_INODE, "server last_rcvd : "LPU64"\n",
 318                le64_to_cpu(fsd->fsd_last_transno));
 319         CDEBUG(D_INODE, "server last_mount: "LPU64"\n",
 320                le64_to_cpu(fsd->fsd_mount_count));
 321
 322         rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off,force_sync);
 323         if (rc)
 324                 CERROR("error writing filter_server_data: rc = %d\n", rc);
 325
 326         RETURN(rc);
 327 }
 328
 329 int filter_update_last_objid(struct obd_device *obd, obd_gr group,
 330                              int force_sync)
 331 {
 332         struct filter_obd *filter = &obd->u.filter;
 333         __u64 tmp;
 334         loff_t off = 0;
 335         int rc;
 336         ENTRY;
 337
 338         if (filter->fo_last_objid_files[group] == NULL) {
 339                 CERROR("Object group "LPU64" not fully setup; not updating "
 340                        "last_objid\n", group);
 341                 RETURN(0);
 342         }
 343
 344         CDEBUG(D_INODE, "server last_objid for group "LPU64": "LPU64"\n",
 345                group, filter->fo_last_objids[group]);
 346
 347         tmp = cpu_to_le64(filter->fo_last_objids[group]);
 348         rc = fsfilt_write_record(obd, filter->fo_last_objid_files[group],
 349                                  &tmp, sizeof(tmp), &off, force_sync);
 350         if (rc)
 351                 CERROR("error writing group "LPU64" last objid: rc = %d\n",
 352                        group, rc);
 353         RETURN(rc);
 354 }
 355
 356 /* assumes caller has already in kernel ctxt */
 357 static int filter_init_server_data(struct obd_device *obd, struct file * filp)
 358 {
 359         struct filter_obd *filter = &obd->u.filter;
 360         struct filter_server_data *fsd;
 361         struct filter_client_data *fcd = NULL;
 362         struct inode *inode = filp->f_dentry->d_inode;
 363         unsigned long last_rcvd_size = inode->i_size;
 364         __u64 mount_count;
 365         int cl_idx;
 366         loff_t off = 0;
 367         int rc;
 368
 369         /* ensure padding in the struct is the correct size */
 370         LASSERT (offsetof(struct filter_server_data, fsd_padding) +
 371                  sizeof(fsd->fsd_padding) == FILTER_LR_SERVER_SIZE);
 372         LASSERT (offsetof(struct filter_client_data, fcd_padding) +
 373                  sizeof(fcd->fcd_padding) == FILTER_LR_CLIENT_SIZE);
 374
 375         OBD_ALLOC(fsd, sizeof(*fsd));
 376         if (!fsd)
 377                 RETURN(-ENOMEM);
 378         filter->fo_fsd = fsd;
 379
 380         OBD_ALLOC(filter->fo_last_rcvd_slots, FILTER_LR_MAX_CLIENTS/8);
 381         if (filter->fo_last_rcvd_slots == NULL) {
 382                 OBD_FREE(fsd, sizeof(*fsd));
 383                 RETURN(-ENOMEM);
 384         }
 385
 386         if (last_rcvd_size == 0) {
 387                 CWARN("%s: initializing new %s\n", obd->obd_name, LAST_RCVD);
 388
 389                 memcpy(fsd->fsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->fsd_uuid));
 390                 fsd->fsd_last_transno = 0;
 391                 mount_count = fsd->fsd_mount_count = 0;
 392                 fsd->fsd_server_size = cpu_to_le32(FILTER_LR_SERVER_SIZE);
 393                 fsd->fsd_client_start = cpu_to_le32(FILTER_LR_CLIENT_START);
 394                 fsd->fsd_client_size = cpu_to_le16(FILTER_LR_CLIENT_SIZE);
 395                 fsd->fsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT);
 396                 filter->fo_subdir_count = FILTER_SUBDIR_COUNT;
 397         } else {
 398                 rc = fsfilt_read_record(obd, filp, fsd, sizeof(*fsd), &off);
 399                 if (rc) {
 400                         CDEBUG(D_INODE,"OBD filter: error reading %s: rc %d\n",
 401                                LAST_RCVD, rc);
 402                         GOTO(err_fsd, rc);
 403                 }
 404                 if (strcmp(fsd->fsd_uuid, obd->obd_uuid.uuid) != 0) {
 405                         CERROR("OBD UUID %s does not match last_rcvd UUID %s\n",
 406                                obd->obd_uuid.uuid, fsd->fsd_uuid);
 407                         GOTO(err_fsd, rc = -EINVAL);
 408                 }
 409                 mount_count = le64_to_cpu(fsd->fsd_mount_count);
 410                 filter->fo_subdir_count = le16_to_cpu(fsd->fsd_subdir_count);
 411         }
 412
 413         if (fsd->fsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) {
 414                 CERROR("unsupported feature %x\n",
 415                        le32_to_cpu(fsd->fsd_feature_incompat) &
 416                        ~FILTER_INCOMPAT_SUPP);
 417                 GOTO(err_fsd, rc = -EINVAL);
 418         }
 419         if (fsd->fsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) {
 420                 CERROR("read-only feature %x\n",
 421                        le32_to_cpu(fsd->fsd_feature_rocompat) &
 422                        ~FILTER_ROCOMPAT_SUPP);
 423                 /* Do something like remount filesystem read-only */
 424                 GOTO(err_fsd, rc = -EINVAL);
 425         }
 426
 427         CDEBUG(D_INODE, "%s: server last_rcvd : "LPU64"\n",
 428                obd->obd_name, le64_to_cpu(fsd->fsd_last_transno));
 429         CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n",
 430                obd->obd_name, mount_count + 1);
 431         CDEBUG(D_INODE, "%s: server data size: %u\n",
 432                obd->obd_name, le32_to_cpu(fsd->fsd_server_size));
 433         CDEBUG(D_INODE, "%s: per-client data start: %u\n",
 434                obd->obd_name, le32_to_cpu(fsd->fsd_client_start));
 435         CDEBUG(D_INODE, "%s: per-client data size: %u\n",
 436                obd->obd_name, le32_to_cpu(fsd->fsd_client_size));
 437         CDEBUG(D_INODE, "%s: server subdir_count: %u\n",
 438                obd->obd_name, le16_to_cpu(fsd->fsd_subdir_count));
 439         CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name,
 440                last_rcvd_size <= le32_to_cpu(fsd->fsd_client_start) ? 0 :
 441                (last_rcvd_size - le32_to_cpu(fsd->fsd_client_start)) /
 442                 le16_to_cpu(fsd->fsd_client_size));
 443
 444         if (!obd->obd_replayable) {
 445                 CWARN("%s: recovery support OFF\n", obd->obd_name);
 446                 GOTO(out, rc = 0);
 447         }
 448
 449         for (cl_idx = 0, off = le32_to_cpu(fsd->fsd_client_start);
 450              off < last_rcvd_size; cl_idx++) {
 451                 __u64 last_rcvd;
 452                 struct obd_export *exp;
 453                 struct filter_export_data *fed;
 454
 455                 if (!fcd) {
 456                         OBD_ALLOC(fcd, sizeof(*fcd));
 457                         if (!fcd)
 458                                 GOTO(err_client, rc = -ENOMEM);
 459                 }
 460
 461                 /* Don't assume off is incremented properly by
 462                  * fsfilt_read_record(), in case sizeof(*fcd)
 463                  * isn't the same as fsd->fsd_client_size.  */
 464                 off = le32_to_cpu(fsd->fsd_client_start) +
 465                         cl_idx * le16_to_cpu(fsd->fsd_client_size);
 466                 rc = fsfilt_read_record(obd, filp, fcd, sizeof(*fcd), &off);
 467                 if (rc) {
 468                         CERROR("error reading FILT %s idx %d off %llu: rc %d\n",
 469                                LAST_RCVD, cl_idx, off, rc);
 470                         break; /* read error shouldn't cause startup to fail */
 471                 }
 472
 473                 if (fcd->fcd_uuid[0] == '\0') {
 474                         CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
 475                                cl_idx);
 476                         continue;
 477                 }
 478
 479                 last_rcvd = le64_to_cpu(fcd->fcd_last_rcvd);
 480
 481                 /* These exports are cleaned up by filter_disconnect(), so they
 482                  * need to be set up like real exports as filter_connect() does.
 483                  */
 484                 exp = class_new_export(obd);
 485                 CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64" "
 486                        "srv lr: "LPU64" fcd_group %d \n", fcd->fcd_uuid, cl_idx,
 487                        last_rcvd, le64_to_cpu(fsd->fsd_last_transno),
 488                        le32_to_cpu(fcd->fcd_group));
 489                 if (exp == NULL)
 490                         GOTO(err_client, rc = -ENOMEM);
 491
 492                 memcpy(&exp->exp_client_uuid.uuid, fcd->fcd_uuid,
 493                        sizeof exp->exp_client_uuid.uuid);
 494                 fed = &exp->exp_filter_data;
 495                 fed->fed_fcd = fcd;
 496                 fed->fed_group = le32_to_cpu(fcd->fcd_group);
 497                 filter_client_add(obd, filter, fed, cl_idx);
 498                 /* create helper if export init gets more complex */
 499                 spin_lock_init(&fed->fed_lock);
 500
 501                 fcd = NULL;
 502                 exp->exp_replay_needed = 1;
 503                 obd->obd_recoverable_clients++;
 504                 obd->obd_max_recoverable_clients++;
 505                 class_export_put(exp);
 506
 507                 CDEBUG(D_OTHER, "client at idx %d has last_rcvd = "LPU64"\n",
 508                        cl_idx, last_rcvd);
 509
 510                 if (last_rcvd > le64_to_cpu(fsd->fsd_last_transno))
 511                         fsd->fsd_last_transno = cpu_to_le64(last_rcvd);
 512
 513         }
 514
 515         obd->obd_last_committed = le64_to_cpu(fsd->fsd_last_transno);
 516
 517         if (obd->obd_recoverable_clients) {
 518                 CWARN("RECOVERY: service %s, %d recoverable clients, "
 519                       "last_transno "LPU64"\n", obd->obd_name,
 520                       obd->obd_recoverable_clients,
 521                       le64_to_cpu(fsd->fsd_last_transno));
 522                 obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
 523                 target_start_recovery_thread(obd, ost_handle);
 524         }
 525
 526         if (fcd)
 527                 OBD_FREE(fcd, sizeof(*fcd));
 528
 529 out:
 530         filter->fo_mount_count = mount_count + 1;
 531         fsd->fsd_mount_count = cpu_to_le64(filter->fo_mount_count);
 532
 533         /* save it, so mount count and last_transno is current */
 534         rc = filter_update_server_data(obd, filp, filter->fo_fsd, 1);
 535         if (rc)
 536                 GOTO(err_client, rc);
 537         RETURN(0);
 538
 539 err_client:
 540         class_disconnect_exports(obd, 0);
 541 err_fsd:
 542         filter_free_server_data(filter);
 543         RETURN(rc);
 544 }
 545
 546 static int filter_cleanup_groups(struct obd_device *obd)
 547 {
 548         struct filter_obd *filter = &obd->u.filter;
 549         struct dentry *dentry;
 550         int i, k;
 551         ENTRY;
 552
 553         for (i = 0; i < filter->fo_group_count; i++) {
 554                 if (filter->fo_subdirs != NULL) {
 555                         for (k = 0; k < filter->fo_subdir_count; k++) {
 556                                 dentry = filter->fo_subdirs[i].dentry[k];
 557                                 if (dentry == NULL)
 558                                         continue;
 559                                 f_dput(dentry);
 560                                 filter->fo_subdirs[i].dentry[k] = NULL;
 561                         }
 562                 }
 563                 if (filter->fo_last_objid_files[i] != NULL) {
 564                         filp_close(filter->fo_last_objid_files[i], 0);
 565                         filter->fo_last_objid_files[i] = NULL;
 566                 }
 567                 if (filter->fo_groups[i] != NULL) {
 568                         dput(filter->fo_groups[i]);
 569                         filter->fo_groups[i] = NULL;
 570                 }
 571         }
 572         if (filter->fo_subdirs != NULL)
 573                 OBD_FREE(filter->fo_subdirs,
 574                          filter->fo_group_count * sizeof(*filter->fo_subdirs));
 575         if (filter->fo_groups != NULL)
 576                 OBD_FREE(filter->fo_groups,
 577                          filter->fo_group_count * sizeof(*filter->fo_groups));
 578         if (filter->fo_last_objids != NULL)
 579                 OBD_FREE(filter->fo_last_objids,
 580                          filter->fo_group_count * sizeof(__u64));
 581         if (filter->fo_last_objid_files != NULL)
 582                 OBD_FREE(filter->fo_last_objid_files,
 583                          filter->fo_group_count * sizeof(struct file *));
 584         f_dput(filter->fo_dentry_O);
 585         RETURN(0);
 586 }
 587
 588 static int filter_update_last_group(struct obd_device *obd, int group)
 589 {
 590         struct filter_obd *filter = &obd->u.filter;
 591         struct file *filp = NULL;
 592         int last_group = 0, rc;
 593         loff_t off = 0;
 594         ENTRY;
 595
 596         if (group <= filter->fo_committed_group)
 597                 RETURN(0);
 598
 599         filp = filp_open("LAST_GROUP", O_RDWR, 0700);
 600         if (IS_ERR(filp)) {
 601                 rc = PTR_ERR(filp);
 602                 filp = NULL;
 603                 CERROR("cannot open LAST_GROUP: rc = %d\n", rc);
 604                 GOTO(cleanup, rc);
 605         }
 606
 607         rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
 608         if (rc) {
 609                 CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n",rc);
 610                 GOTO(cleanup, rc);
 611         }
 612         LASSERT(off == 0 || last_group >= FILTER_MIN_GROUPS);
 613         CDEBUG(D_INODE, "%s: previous %d, new %d\n",
 614                obd->obd_name, last_group, group);
 615
 616         off = 0;
 617         last_group = group;
 618         /* must be sync: bXXXX */
 619         rc = fsfilt_write_record(obd, filp, &last_group, sizeof(__u32), &off, 1);
 620         if (rc) {
 621                 CDEBUG(D_INODE, "error updating LAST_GROUP: rc %d\n", rc);
 622                 GOTO(cleanup, rc);
 623         }
 624
 625         filter->fo_committed_group = group;
 626 cleanup:
 627         if (filp)
 628                 filp_close(filp, 0);
 629         RETURN(rc);
 630 }
 631
 632 static int filter_read_group_internal(struct obd_device *obd, int group,
 633                                       int create)
 634 {
 635         struct filter_obd *filter = &obd->u.filter;
 636         __u64 *new_objids = NULL;
 637         struct filter_subdirs *new_subdirs = NULL, *tmp_subdirs = NULL;
 638         struct dentry **new_groups = NULL;
 639         struct file **new_files = NULL;
 640         struct dentry *dentry;
 641         struct file *filp;
 642         int old_count = filter->fo_group_count, rc, stage = 0, i;
 643         char name[25];
 644         __u64 last_objid;
 645         loff_t off = 0;
 646
 647         snprintf(name, 24, "%d", group);
 648         name[24] = '\0';
 649
 650         if (!create) {
 651                 dentry = ll_lookup_one_len(name, filter->fo_dentry_O,
 652                                            strlen(name));
 653                 if (IS_ERR(dentry)) {
 654                         CERROR("Cannot lookup expected object group %d: %ld\n",
 655                                group, PTR_ERR(dentry));
 656                         RETURN(PTR_ERR(dentry));
 657                 }
 658         } else {
 659                 dentry = simple_mkdir(filter->fo_dentry_O, name, 0700, 1);
 660                 if (IS_ERR(dentry)) {
 661                         CERROR("cannot lookup/create O/%s: rc = %ld\n", name,
 662                                PTR_ERR(dentry));
 663                         RETURN(PTR_ERR(dentry));
 664                 }
 665         }
 666         stage = 1;
 667
 668         snprintf(name, 24, "O/%d/LAST_ID", group);
 669         name[24] = '\0';
 670         filp = filp_open(name, O_CREAT | O_RDWR, 0700);
 671         if (IS_ERR(filp)) {
 672                 CERROR("cannot create %s: rc = %ld\n", name, PTR_ERR(filp));
 673                 GOTO(cleanup, rc = PTR_ERR(filp));
 674         }
 675         stage = 2;
 676
 677         rc = fsfilt_read_record(obd, filp, &last_objid, sizeof(__u64), &off);
 678         if (rc) {
 679                 CDEBUG(D_INODE, "error reading %s: rc %d\n", name, rc);
 680                 GOTO(cleanup, rc);
 681         }
 682
 683         if (filter->fo_subdir_count) {
 684                 OBD_ALLOC(tmp_subdirs, sizeof(*tmp_subdirs));
 685                 if (tmp_subdirs == NULL)
 686                         GOTO(cleanup, rc = -ENOMEM);
 687                 stage = 3;
 688
 689                 for (i = 0; i < filter->fo_subdir_count; i++) {
 690                         char dir[20];
 691                         snprintf(dir, sizeof(dir), "d%u", i);
 692
 693                         tmp_subdirs->dentry[i] =
 694                                 simple_mkdir(dentry, dir, 0700, 1);
 695                         if (IS_ERR(tmp_subdirs->dentry[i])) {
 696                                 rc = PTR_ERR(tmp_subdirs->dentry[i]);
 697                                 CERROR("can't lookup/create O/%d/%s: rc = %d\n",
 698                                        group, dir, rc);
 699                                 GOTO(cleanup, rc);
 700                         }
 701                         CDEBUG(D_INODE, "got/created O/%d/%s: %p\n", group, dir,
 702                                tmp_subdirs->dentry[i]);
 703                 }
 704         }
 705
 706         /* 'group' is an index; we need an array of length 'group + 1' */
 707         if (group + 1 > old_count) {
 708                 int len = group + 1;
 709                 OBD_ALLOC(new_objids, len * sizeof(*new_objids));
 710                 OBD_ALLOC(new_subdirs, len * sizeof(*new_subdirs));
 711                 OBD_ALLOC(new_groups, len * sizeof(*new_groups));
 712                 OBD_ALLOC(new_files, len * sizeof(*new_files));
 713                 stage = 4;
 714                 if (new_objids == NULL || new_subdirs == NULL ||
 715                     new_groups == NULL || new_files == NULL)
 716                         GOTO(cleanup, rc = -ENOMEM);
 717
 718                 memcpy(new_objids, filter->fo_last_objids,
 719                        old_count * sizeof(*new_objids));
 720                 memcpy(new_subdirs, filter->fo_subdirs,
 721                        old_count * sizeof(*new_subdirs));
 722                 memcpy(new_groups, filter->fo_groups,
 723                        old_count * sizeof(*new_groups));
 724                 memcpy(new_files, filter->fo_last_objid_files,
 725                        old_count * sizeof(*new_files));
 726
 727                 if (old_count) {
 728                         OBD_FREE(filter->fo_last_objids,
 729                                  old_count * sizeof(*new_objids));
 730                         OBD_FREE(filter->fo_subdirs,
 731                                  old_count * sizeof(*new_subdirs));
 732                         OBD_FREE(filter->fo_groups,
 733                                  old_count * sizeof(*new_groups));
 734                         OBD_FREE(filter->fo_last_objid_files,
 735                                  old_count * sizeof(*new_files));
 736                 }
 737                 filter->fo_last_objids = new_objids;
 738                 filter->fo_subdirs = new_subdirs;
 739                 filter->fo_groups = new_groups;
 740                 filter->fo_last_objid_files = new_files;
 741                 filter->fo_group_count = len;
 742         }
 743
 744         filter->fo_groups[group] = dentry;
 745         filter->fo_last_objid_files[group] = filp;
 746         if (filter->fo_subdir_count) {
 747                 filter->fo_subdirs[group] = *tmp_subdirs;
 748                 OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
 749         }
 750
 751         filter_update_last_group(obd, group);
 752
 753         if (filp->f_dentry->d_inode->i_size == 0) {
 754                 filter->fo_last_objids[group] = FILTER_INIT_OBJID;
 755                 RETURN(0);
 756         }
 757
 758         filter->fo_last_objids[group] = le64_to_cpu(last_objid);
 759         CDEBUG(D_INODE, "%s: server last_objid group %d: "LPU64"\n",
 760                obd->obd_name, group, last_objid);
 761         RETURN(0);
 762  cleanup:
 763         switch (stage) {
 764         case 4:
 765                 if (new_objids != NULL)
 766                         OBD_FREE(new_objids, group * sizeof(*new_objids));
 767                 if (new_subdirs != NULL)
 768                         OBD_FREE(new_subdirs, group * sizeof(*new_subdirs));
 769                 if (new_groups != NULL)
 770                         OBD_FREE(new_groups, group * sizeof(*new_groups));
 771                 if (new_files != NULL)
 772                         OBD_FREE(new_files, group * sizeof(*new_files));
 773         case 3:
 774                 if (filter->fo_subdir_count) {
 775                         for (i = 0; i < filter->fo_subdir_count; i++) {
 776                                 if (tmp_subdirs->dentry[i] != NULL)
 777                                         dput(tmp_subdirs->dentry[i]);
 778                         }
 779                         OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
 780                 }
 781         case 2:
 782                 filp_close(filp, 0);
 783         case 1:
 784                 dput(dentry);
 785         }
 786         RETURN(rc);
 787 }
 788
 789 static int filter_read_groups(struct obd_device *obd, int last_group,
 790                               int create)
 791 {
 792         struct filter_obd *filter = &obd->u.filter;
 793         int old_count, group, rc = 0;
 794
 795         down(&filter->fo_init_lock);
 796         old_count = filter->fo_group_count;
 797         for (group = old_count; group <= last_group; group++) {
 798                 if (group == 0)
 799                         continue; /* no group zero */
 800
 801                 rc = filter_read_group_internal(obd, group, create);
 802                 if (rc != 0)
 803                         break;
 804         }
 805         up(&filter->fo_init_lock);
 806         return rc;
 807 }
 808
 809 static int filter_prep_groups(struct obd_device *obd)
 810 {
 811         struct filter_obd *filter = &obd->u.filter;
 812         struct dentry *dentry, *O_dentry;
 813         int rc = 0, cleanup_phase = 0;
 814         struct file *filp = NULL;
 815         int last_group;
 816         loff_t off = 0;
 817         ENTRY;
 818
 819         O_dentry = simple_mkdir(current->fs->pwd, "O", 0700, 1);
 820         CDEBUG(D_INODE, "got/created O: %p\n", O_dentry);
 821         if (IS_ERR(O_dentry)) {
 822                 rc = PTR_ERR(O_dentry);
 823                 CERROR("cannot open/create O: rc = %d\n", rc);
 824                 GOTO(cleanup, rc);
 825         }
 826         filter->fo_dentry_O = O_dentry;
 827         cleanup_phase = 1; /* O_dentry */
 828
 829         /* Lookup "R" to tell if we're on an old OST FS and need to convert
 830          * from O/R/<dir>/<objid> to O/0/<dir>/<objid>.  This can be removed
 831          * some time post 1.0 when all old-style OSTs have converted along
 832          * with the init_objid hack. */
 833         dentry = ll_lookup_one_len("R", O_dentry, 1);
 834         if (IS_ERR(dentry))
 835                 GOTO(cleanup, rc = PTR_ERR(dentry));
 836         if (dentry->d_inode && S_ISDIR(dentry->d_inode->i_mode)) {
 837                 struct dentry *O0_dentry = lookup_one_len("0", O_dentry, 1);
 838                 ENTRY;
 839
 840                 CWARN("converting OST to new object layout\n");
 841                 if (IS_ERR(O0_dentry)) {
 842                         rc = PTR_ERR(O0_dentry);
 843                         CERROR("error looking up O/0: rc %d\n", rc);
 844                         GOTO(cleanup_R, rc);
 845                 }
 846
 847                 if (O0_dentry->d_inode) {
 848                         CERROR("Both O/R and O/0 exist. Fix manually.\n");
 849                         GOTO(cleanup_O0, rc = -EEXIST);
 850                 }
 851
 852                 down(&O_dentry->d_inode->i_sem);
 853                 rc = vfs_rename(O_dentry->d_inode, dentry,
 854                                 O_dentry->d_inode, O0_dentry);
 855                 up(&O_dentry->d_inode->i_sem);
 856
 857                 if (rc) {
 858                         CERROR("error renaming O/R to O/0: rc %d\n", rc);
 859                         GOTO(cleanup_O0, rc);
 860                 }
 861                 filter->fo_fsd->fsd_feature_incompat |=
 862                         cpu_to_le32(FILTER_INCOMPAT_GROUPS);
 863                 rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
 864                                                filter->fo_fsd, 1);
 865                 GOTO(cleanup_O0, rc);
 866
 867         cleanup_O0:
 868                 f_dput(O0_dentry);
 869         cleanup_R:
 870                 f_dput(dentry);
 871                 if (rc)
 872                         GOTO(cleanup, rc);
 873         } else {
 874                 f_dput(dentry);
 875         }
 876
 877         cleanup_phase = 2; /* groups */
 878
 879         /* we have to initialize all groups before first connections from
 880          * clients because they may send create/destroy for any group -bzzz */
 881         filp = filp_open("LAST_GROUP", O_CREAT | O_RDWR, 0700);
 882         if (IS_ERR(filp)) {
 883                 CERROR("cannot create LAST_GROUP: rc = %ld\n", PTR_ERR(filp));
 884                 GOTO(cleanup, rc = PTR_ERR(filp));
 885         }
 886         cleanup_phase = 3; /* filp */
 887
 888         rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
 889         if (rc) {
 890                 CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n", rc);
 891                 GOTO(cleanup, rc);
 892         }
 893         if (off == 0) {
 894                 last_group = FILTER_MIN_GROUPS;
 895         } else {
 896                 LASSERT(last_group >= FILTER_MIN_GROUPS);
 897         }
 898
 899         CWARN("%s: initialize groups [%d,%d]\n", obd->obd_name,
 900               FILTER_MIN_GROUPS, last_group);
 901         filter->fo_committed_group = last_group;
 902         rc = filter_read_groups(obd, last_group, 1);
 903         if (rc)
 904                 GOTO(cleanup, rc);
 905
 906         filp_close(filp, 0);
 907         RETURN(0);
 908
 909  cleanup:
 910         switch (cleanup_phase) {
 911         case 3:
 912                 filp_close(filp, 0);
 913         case 2:
 914                 filter_cleanup_groups(obd);
 915         case 1:
 916                 f_dput(filter->fo_dentry_O);
 917                 filter->fo_dentry_O = NULL;
 918         default:
 919                 break;
 920         }
 921         return rc;
 922 }
 923
 924 /* setup the object store with correct subdirectories */
 925 static int filter_prep(struct obd_device *obd)
 926 {
 927         struct lvfs_run_ctxt saved;
 928         struct filter_obd *filter = &obd->u.filter;
 929         struct file *file;
 930         struct inode *inode;
 931         int rc = 0;
 932         ENTRY;
 933
 934         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 935         file = filp_open(LAST_RCVD, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
 936         if (!file || IS_ERR(file)) {
 937                 rc = PTR_ERR(file);
 938                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
 939                        LAST_RCVD, rc);
 940                 GOTO(out, rc);
 941         }
 942
 943         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
 944                 CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD,
 945                        file->f_dentry->d_inode->i_mode);
 946                 GOTO(err_filp, rc = -ENOENT);
 947         }
 948
 949         /* steal operations */
 950         inode = file->f_dentry->d_inode;
 951         filter->fo_fop = file->f_op;
 952         filter->fo_iop = inode->i_op;
 953         filter->fo_aops = inode->i_mapping->a_ops;
 954
 955         rc = filter_init_server_data(obd, file);
 956         if (rc) {
 957                 CERROR("cannot read %s: rc = %d\n", LAST_RCVD, rc);
 958                 GOTO(err_filp, rc);
 959         }
 960         filter->fo_rcvd_filp = file;
 961
 962         rc = filter_prep_groups(obd);
 963         if (rc)
 964                 GOTO(err_server_data, rc);
 965
 966  out:
 967         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 968
 969         return(rc);
 970
 971  err_server_data:
 972         //class_disconnect_exports(obd, 0);
 973         filter_free_server_data(filter);
 974  err_filp:
 975         if (filp_close(file, 0))
 976                 CERROR("can't close %s after error\n", LAST_RCVD);
 977         filter->fo_rcvd_filp = NULL;
 978         goto out;
 979 }
 980
 981 /* cleanup the filter: write last used object id to status file */
 982 static void filter_post(struct obd_device *obd)
 983 {
 984         struct lvfs_run_ctxt saved;
 985         struct filter_obd *filter = &obd->u.filter;
 986         int rc, i;
 987
 988         /* XXX: filter_update_lastobjid used to call fsync_dev.  It might be
 989          * best to start a transaction with h_sync, because we removed this
 990          * from lastobjid */
 991
 992         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 993         rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
 994                                        filter->fo_fsd, 0);
 995         if (rc)
 996                 CERROR("error writing server data: rc = %d\n", rc);
 997
 998         for (i = 1; i < filter->fo_group_count; i++) {
 999                 rc = filter_update_last_objid(obd, i,
1000                                              (i == filter->fo_group_count - 1));
1001                 if (rc)
1002                         CERROR("error writing group %d lastobjid: rc = %d\n",
1003                                i, rc);
1004         }
1005
1006         rc = filp_close(filter->fo_rcvd_filp, 0);
1007         filter->fo_rcvd_filp = NULL;
1008         if (rc)
1009                 CERROR("error closing %s: rc = %d\n", LAST_RCVD, rc);
1010
1011         filter_cleanup_groups(obd);
1012         filter_free_server_data(filter);
1013         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1014 }
1015
1016 static void filter_set_last_id(struct filter_obd *filter, int group, obd_id id)
1017 {
1018         LASSERT(filter->fo_fsd != NULL);
1019         LASSERT(group > 0);
1020         LASSERT(group < filter->fo_group_count);
1021
1022         spin_lock(&filter->fo_objidlock);
1023         filter->fo_last_objids[group] = id;
1024         spin_unlock(&filter->fo_objidlock);
1025 }
1026
1027 __u64 filter_last_id(struct filter_obd *filter, int group)
1028 {
1029         obd_id id;
1030         LASSERT(filter->fo_fsd != NULL);
1031         LASSERT(group > 0);
1032         LASSERT(group < filter->fo_group_count);
1033
1034         spin_lock(&filter->fo_objidlock);
1035         id = filter->fo_last_objids[group];
1036         spin_unlock(&filter->fo_objidlock);
1037
1038         return id;
1039 }
1040
1041 /* direct cut-n-paste of mds_blocking_ast() */
1042 static int filter_blocking_ast(struct ldlm_lock *lock,
1043                                struct ldlm_lock_desc *desc,
1044                                void *data, int flag)
1045 {
1046         int do_ast;
1047         ENTRY;
1048
1049         if (flag == LDLM_CB_CANCELING) {
1050                 /* Don't need to do anything here. */
1051                 RETURN(0);
1052         }
1053
1054         /* XXX layering violation!  -phil */
1055         l_lock(&lock->l_resource->lr_namespace->ns_lock);
1056         /* Get this: if filter_blocking_ast is racing with ldlm_intent_policy,
1057          * such that filter_blocking_ast is called just before l_i_p takes the
1058          * ns_lock, then by the time we get the lock, we might not be the
1059          * correct blocking function anymore.  So check, and return early, if
1060          * so. */
1061         if (lock->l_blocking_ast != filter_blocking_ast) {
1062                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
1063                 RETURN(0);
1064         }
1065
1066         lock->l_flags |= LDLM_FL_CBPENDING;
1067         do_ast = (!lock->l_readers && !lock->l_writers);
1068         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
1069
1070         if (do_ast) {
1071                 struct lustre_handle lockh;
1072                 int rc;
1073
1074                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
1075                 ldlm_lock2handle(lock, &lockh);
1076                 rc = ldlm_cli_cancel(&lockh);
1077                 if (rc < 0)
1078                         CERROR("ldlm_cli_cancel: %d\n", rc);
1079         } else {
1080                 LDLM_DEBUG(lock, "Lock still has references, will be "
1081                            "cancelled later");
1082         }
1083         RETURN(0);
1084 }
1085
1086 extern void *lock_dir(struct inode *dir, struct qstr *name);
1087 extern void unlock_dir(struct inode *dir, void *lock);
1088
1089 static void * filter_lock_dentry(struct obd_device *obd,
1090                                  struct dentry *dparent, obd_id id)
1091 {
1092 #ifdef S_PDIROPS
1093         struct qstr qstr;
1094         char name[32];
1095         qstr.name = name;
1096         qstr.len = sprintf(name, LPU64, id);
1097         return lock_dir(dparent->d_inode, &qstr);
1098 #else
1099         down(&dparent->d_inode->i_sem);
1100 #endif
1101         return 0;
1102 }
1103
1104 /* We never dget the object parent, so DON'T dput it either */
1105 static void filter_parent_unlock(struct dentry *dparent, void *lock)
1106 {
1107 #ifdef S_PDIROPS
1108         LASSERT(lock != NULL);
1109         unlock_dir(dparent->d_inode, lock);
1110 #else
1111         up(&dparent->d_inode->i_sem);
1112 #endif
1113 }
1114
1115 /* We never dget the object parent, so DON'T dput it either */
1116 struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid)
1117 {
1118         struct filter_obd *filter = &obd->u.filter;
1119         LASSERT(group < filter->fo_group_count);
1120         LASSERT(group > 0);
1121
1122         if (filter->fo_subdir_count == 0)
1123                 return filter->fo_groups[group];
1124
1125         return filter->fo_subdirs[group].dentry[objid & (filter->fo_subdir_count - 1)];
1126 }
1127
1128 /* We never dget the object parent, so DON'T dput it either */
1129 struct dentry *filter_parent_lock(struct obd_device *obd, obd_gr group,
1130                                   obd_id objid, void **lock)
1131 {
1132         unsigned long now = jiffies;
1133         struct dentry *dparent = filter_parent(obd, group, objid);
1134
1135         if (IS_ERR(dparent))
1136                 return dparent;
1137
1138         LASSERT(dparent);
1139         LASSERT(dparent->d_inode);
1140
1141         *lock = filter_lock_dentry(obd, dparent, objid);
1142         fsfilt_check_slow(now, obd_timeout, "parent lock");
1143         return dparent;
1144 }
1145
1146 /* How to get files, dentries, inodes from object id's.
1147  *
1148  * If dir_dentry is passed, the caller has already locked the parent
1149  * appropriately for this operation (normally a write lock).  If
1150  * dir_dentry is NULL, we do a read lock while we do the lookup to
1151  * avoid races with create/destroy and such changing the directory
1152  * internal to the filesystem code. */
1153 struct dentry *filter_id2dentry(struct obd_device *obd,
1154                                 struct dentry *dir_dentry,
1155                                 obd_gr group, obd_id id)
1156 {
1157         struct dentry *dparent = dir_dentry;
1158         struct dentry *dchild;
1159         void *lock = NULL;
1160         char name[32];
1161         int len;
1162         ENTRY;
1163
1164         if (id == 0) {
1165                 CERROR("fatal: invalid object id 0\n");
1166                 RETURN(ERR_PTR(-ESTALE));
1167         }
1168
1169         len = sprintf(name, LPU64, id);
1170         if (dir_dentry == NULL) {
1171                 dparent = filter_parent_lock(obd, group, id, &lock);
1172                 if (IS_ERR(dparent)) {
1173                         CERROR("%s: error getting object "LPU64":"LPU64
1174                                " parent: rc %ld\n", obd->obd_name,
1175                                id, group, PTR_ERR(dparent));
1176                         RETURN(dparent);
1177                 }
1178         }
1179         CDEBUG(D_INODE, "looking up object O/%.*s/%s\n",
1180                dparent->d_name.len, dparent->d_name.name, name);
1181         dchild = /*ll_*/lookup_one_len(name, dparent, len);
1182         if (dir_dentry == NULL)
1183                 filter_parent_unlock(dparent, lock);
1184         if (IS_ERR(dchild)) {
1185                 CERROR("%s: child lookup error %ld\n", obd->obd_name,
1186                        PTR_ERR(dchild));
1187                 RETURN(dchild);
1188         }
1189
1190         if (dchild->d_inode != NULL && is_bad_inode(dchild->d_inode)) {
1191                 CERROR("%s: got bad inode "LPU64"\n", obd->obd_name, id);
1192                 f_dput(dchild);
1193                 RETURN(ERR_PTR(-ENOENT));
1194         }
1195
1196         CDEBUG(D_INODE, "got child objid %s: %p, count = %d\n",
1197                name, dchild, atomic_read(&dchild->d_count));
1198
1199         LASSERT(atomic_read(&dchild->d_count) > 0);
1200
1201         RETURN(dchild);
1202 }
1203
1204 static int filter_prepare_destroy(struct obd_device *obd, obd_id objid,
1205                                   obd_id group)
1206 {
1207         struct lustre_handle lockh;
1208         int flags = LDLM_AST_DISCARD_DATA, rc;
1209         struct ldlm_res_id res_id = { .name = { objid, 0, group, 0 } };
1210         ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
1211
1212         ENTRY;
1213         /* Tell the clients that the object is gone now and that they should
1214          * throw away any cached pages. */
1215         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, res_id,
1216                               LDLM_EXTENT, &policy, LCK_PW,
1217                               &flags, filter_blocking_ast, ldlm_completion_ast,
1218                               NULL, NULL, NULL, 0, NULL, &lockh);
1219
1220         /* We only care about the side-effects, just drop the lock. */
1221         if (rc == ELDLM_OK)
1222                 ldlm_lock_decref(&lockh, LCK_PW);
1223
1224         RETURN(rc);
1225 }
1226
1227 /* Caller must hold LCK_PW on parent and push us into kernel context.
1228  * Caller is also required to ensure that dchild->d_inode exists. */
1229 static int filter_destroy_internal(struct obd_device *obd, obd_id objid,
1230                                    struct dentry *dparent,
1231                                    struct dentry *dchild)
1232 {
1233         struct inode *inode = dchild->d_inode;
1234         int rc;
1235         ENTRY;
1236
1237         if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) {
1238                 CERROR("destroying objid %.*s nlink = %lu, count = %d\n",
1239                        dchild->d_name.len, dchild->d_name.name,
1240                        (unsigned long)inode->i_nlink,
1241                        atomic_read(&inode->i_count));
1242         }
1243
1244         rc = vfs_unlink(dparent->d_inode, dchild);
1245
1246         if (rc)
1247                 CERROR("error unlinking objid %.*s: rc %d\n",
1248                        dchild->d_name.len, dchild->d_name.name, rc);
1249
1250         RETURN(rc);
1251 }
1252
1253 static int filter_intent_policy(struct ldlm_namespace *ns,
1254                                 struct ldlm_lock **lockp, void *req_cookie,
1255                                 ldlm_mode_t mode, int flags, void *data)
1256 {
1257         struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
1258         struct ptlrpc_request *req = req_cookie;
1259         struct ldlm_lock *lock = *lockp, *l = NULL;
1260         struct ldlm_resource *res = lock->l_resource;
1261         ldlm_processing_policy policy;
1262         struct ost_lvb *res_lvb, *reply_lvb;
1263         struct ldlm_reply *rep;
1264         struct list_head *tmp;
1265         ldlm_error_t err;
1266         int tmpflags = 0, rc, repsize[2] = {sizeof(*rep), sizeof(*reply_lvb)};
1267         ENTRY;
1268
1269         policy = ldlm_get_processing_policy(res);
1270         LASSERT(policy != NULL);
1271         LASSERT(req != NULL);
1272
1273         rc = lustre_pack_reply(req, 2, repsize, NULL);
1274         if (rc)
1275                 RETURN(req->rq_status = rc);
1276
1277         rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*rep));
1278         LASSERT(rep != NULL);
1279
1280         reply_lvb = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*reply_lvb));
1281         LASSERT(reply_lvb != NULL);
1282
1283         //fixup_handle_for_resent_req(req, lock, &lockh);
1284
1285         /* If we grant any lock at all, it will be a whole-file read lock.
1286          * Call the extent policy function to see if our request can be
1287          * granted, or is blocked. */
1288         lock->l_policy_data.l_extent.start = 0;
1289         lock->l_policy_data.l_extent.end = OBD_OBJECT_EOF;
1290         lock->l_req_mode = LCK_PR;
1291
1292         l_lock(&res->lr_namespace->ns_lock);
1293
1294         res->lr_tmp = &rpc_list;
1295         rc = policy(lock, &tmpflags, 0, &err);
1296         res->lr_tmp = NULL;
1297
1298         /* FIXME: we should change the policy function slightly, to not make
1299          * this list at all, since we just turn around and free it */
1300         while (!list_empty(&rpc_list)) {
1301                 struct ldlm_ast_work *w =
1302                         list_entry(rpc_list.next, struct ldlm_ast_work, w_list);
1303                 list_del(&w->w_list);
1304                 LDLM_LOCK_PUT(w->w_lock);
1305                 OBD_FREE(w, sizeof(*w));
1306         }
1307
1308         if (rc == LDLM_ITER_CONTINUE) {
1309                 /* The lock met with no resistance; we're finished. */
1310                 l_unlock(&res->lr_namespace->ns_lock);
1311                 RETURN(ELDLM_LOCK_REPLACED);
1312         }
1313
1314         /* Do not grant any lock, but instead send GL callbacks.  The extent
1315          * policy nicely created a list of all PW locks for us.  We will choose
1316          * the highest of those which are larger than the size in the LVB, if
1317          * any, and perform a glimpse callback. */
1318         down(&res->lr_lvb_sem);
1319         res_lvb = res->lr_lvb_data;
1320         LASSERT(res_lvb != NULL);
1321         *reply_lvb = *res_lvb;
1322         up(&res->lr_lvb_sem);
1323
1324         list_for_each(tmp, &res->lr_granted) {
1325                 struct ldlm_lock *tmplock =
1326                         list_entry(tmp, struct ldlm_lock, l_res_link);
1327
1328                 if (tmplock->l_granted_mode == LCK_PR)
1329                         continue;
1330
1331                 if (tmplock->l_policy_data.l_extent.end <= reply_lvb->lvb_size)
1332                         continue;
1333
1334                 if (l == NULL) {
1335                         l = LDLM_LOCK_GET(tmplock);
1336                         continue;
1337                 }
1338
1339                 if (l->l_policy_data.l_extent.start >
1340                     tmplock->l_policy_data.l_extent.start)
1341                         continue;
1342
1343                 LDLM_LOCK_PUT(l);
1344                 l = LDLM_LOCK_GET(tmplock);
1345         }
1346         l_unlock(&res->lr_namespace->ns_lock);
1347
1348         /* There were no PW locks beyond the size in the LVB; finished. */
1349         if (l == NULL)
1350                 RETURN(ELDLM_LOCK_ABORTED);
1351
1352         if (l->l_glimpse_ast == NULL) {
1353                 /* We are racing with unlink(); just return -ENOENT */
1354                 rep->lock_policy_res1 = -ENOENT;
1355                 goto out;
1356         }
1357
1358         LASSERTF(l->l_glimpse_ast != NULL, "l == %p", l);
1359
1360         rc = l->l_glimpse_ast(l, NULL); /* this will update the LVB */
1361         if (rc != 0 && res->lr_namespace->ns_lvbo &&
1362             res->lr_namespace->ns_lvbo->lvbo_update) {
1363                 res->lr_namespace->ns_lvbo->lvbo_update(res, NULL, 0, 1);
1364         }
1365
1366         down(&res->lr_lvb_sem);
1367         *reply_lvb = *res_lvb;
1368         up(&res->lr_lvb_sem);
1369 out:
1370         LDLM_LOCK_PUT(l);
1371
1372         RETURN(ELDLM_LOCK_ABORTED);
1373 }
1374
1375 static int filter_post_fs_cleanup(struct obd_device *obd)
1376 {
1377         int rc = 0;
1378
1379         rc = fsfilt_post_cleanup(obd);
1380
1381         RETURN(rc);
1382 }
1383
1384 static int filter_group_set_fs_flags(struct obd_device *obd, int group)
1385 {
1386         struct filter_obd *filter = &obd->u.filter;
1387         int rc = 0, i = 0;
1388         ENTRY;
1389
1390         /* zero group is not longer valid. */
1391         if (group== 0)
1392                 RETURN(rc);
1393         for (i = 0; i < filter->fo_subdir_count; i++) {
1394                 struct dentry *dentry;
1395                 dentry = (filter->fo_subdirs + group)->dentry[i];
1396                 rc = fsfilt_set_fs_flags(obd, dentry->d_inode,
1397                                          SM_DO_REC | SM_DO_COW);
1398                 if (rc)
1399                         RETURN(rc);
1400         }
1401         RETURN(rc);
1402 }
1403 static int filter_post_fs_setup(struct obd_device *obd)
1404 {
1405         struct filter_obd *filter = &obd->u.filter;
1406         int rc = 0, j = 0;
1407         struct llog_ctxt *ctxt = NULL;
1408
1409         rc = fsfilt_post_setup(obd, filter->fo_dentry_O);
1410         if (rc)
1411                 RETURN(rc);
1412
1413         for (j = 0; j < filter->fo_group_count; j++) {
1414                 rc = filter_group_set_fs_flags(obd, j);
1415                 if (rc)
1416                         return rc;
1417         }
1418
1419         fsfilt_get_reint_log_ctxt(obd, filter->fo_sb, &ctxt);
1420         if (ctxt) {
1421                 ctxt->loc_obd = obd;
1422                 ctxt->loc_idx = LLOG_REINT_ORIG_CTXT;
1423                 obd->obd_llog_ctxt[LLOG_REINT_ORIG_CTXT] = ctxt;
1424         }
1425         fsfilt_set_ost_flags(obd, filter->fo_sb);
1426         return rc;
1427 }
1428
1429 /* mount the file system (secretly) */
1430 int filter_common_setup(struct obd_device *obd, obd_count len, void *buf,
1431                         char *option)
1432 {
1433         struct lustre_cfg* lcfg = buf;
1434         struct filter_obd *filter = &obd->u.filter;
1435         struct lvfs_obd_ctxt *lvfs_ctxt = NULL;
1436         struct vfsmount *mnt;
1437         char ns_name[48];
1438         int rc = 0, i;
1439         ENTRY;
1440
1441
1442         if (!lcfg->lcfg_inlbuf1 || !lcfg->lcfg_inlbuf2)
1443                 RETURN(-EINVAL);
1444
1445         obd->obd_fsops = fsfilt_get_ops(lcfg->lcfg_inlbuf2);
1446         if (IS_ERR(obd->obd_fsops))
1447                 RETURN(PTR_ERR(obd->obd_fsops));
1448
1449         rc = lvfs_mount_fs(lcfg->lcfg_inlbuf1, lcfg->lcfg_inlbuf2,
1450                            option, MS_NOATIME | MS_NODIRATIME, &lvfs_ctxt);
1451         if (rc) {
1452                 CERROR("lvfs_mount_fs failed: rc = %d\n", rc);
1453                 GOTO(err_ops, rc);
1454         }
1455         LASSERT(lvfs_ctxt);
1456
1457         mnt = lvfs_ctxt->loc_mnt;
1458         filter->fo_lvfs_ctxt = lvfs_ctxt;
1459
1460         if (lcfg->lcfg_inllen3 > 0 && lcfg->lcfg_inlbuf3) {
1461                 if (*lcfg->lcfg_inlbuf3 == 'f') {
1462                         obd->obd_replayable = 1;
1463                         obd_sync_filter = 1;
1464                         CWARN("%s: recovery enabled\n", obd->obd_name);
1465                 } else {
1466                         if (*lcfg->lcfg_inlbuf3 != 'n') {
1467                                 CERROR("unrecognised flag '%c'\n",
1468                                        *lcfg->lcfg_inlbuf3);
1469                         }
1470                         // XXX Robert? Why do we get errors here
1471                         // GOTO(err_mntput, rc = -EINVAL);
1472                 }
1473         }
1474
1475         filter->fo_vfsmnt = mnt;
1476         filter->fo_sb = mnt->mnt_sb;
1477         filter->fo_fstype = mnt->mnt_sb->s_type->name;
1478         CDEBUG(D_SUPER, "%s: mnt = %p\n", filter->fo_fstype, mnt);
1479
1480         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1481         obd->obd_lvfs_ctxt.pwdmnt = mnt;
1482         obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
1483         obd->obd_lvfs_ctxt.fs = get_ds();
1484         obd->obd_lvfs_ctxt.cb_ops = filter_lvfs_ops;
1485
1486         ll_clear_rdonly(ll_sbdev(filter->fo_sb));
1487
1488         rc = fsfilt_setup(obd, mnt->mnt_sb);
1489         if (rc)
1490                 GOTO(err_mntput, rc);
1491
1492         sema_init(&filter->fo_init_lock, 1);
1493         filter->fo_committed_group = 0;
1494         rc = filter_prep(obd);
1495         if (rc)
1496                 GOTO(err_mntput, rc);
1497
1498         filter->fo_destroys_in_progress = 0;
1499         for (i = 0; i < 32; i++)
1500                 sema_init(&filter->fo_create_locks[i], 1);
1501
1502         spin_lock_init(&filter->fo_translock);
1503         spin_lock_init(&filter->fo_objidlock);
1504         INIT_LIST_HEAD(&filter->fo_export_list);
1505         sema_init(&filter->fo_alloc_lock, 1);
1506         spin_lock_init(&filter->fo_r_pages.oh_lock);
1507         spin_lock_init(&filter->fo_w_pages.oh_lock);
1508         spin_lock_init(&filter->fo_r_discont_pages.oh_lock);
1509         spin_lock_init(&filter->fo_w_discont_pages.oh_lock);
1510         spin_lock_init(&filter->fo_r_discont_blocks.oh_lock);
1511         spin_lock_init(&filter->fo_w_discont_blocks.oh_lock);
1512         filter->fo_readcache_max_filesize = FILTER_MAX_CACHE_SIZE;
1513
1514         INIT_LIST_HEAD(&filter->fo_llog_list);
1515         spin_lock_init(&filter->fo_llog_list_lock);
1516
1517         sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid);
1518         obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER);
1519
1520         if (obd->obd_namespace == NULL)
1521                 GOTO(err_post, rc = -ENOMEM);
1522         obd->obd_namespace->ns_lvbp = obd;
1523         obd->obd_namespace->ns_lvbo = &filter_lvbo;
1524         ldlm_register_intent(obd->obd_namespace, filter_intent_policy);
1525
1526         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
1527                            "filter_ldlm_cb_client", &obd->obd_ldlm_client);
1528
1529         rc = obd_llog_cat_initialize(obd, &obd->obd_llogs, 1, CATLIST);
1530         if (rc) {
1531                 CERROR("failed to setup llogging subsystems\n");
1532                 GOTO(err_post, rc);
1533         }
1534         RETURN(0);
1535
1536 err_post:
1537         filter_post(obd);
1538 err_mntput:
1539         unlock_kernel();
1540         lvfs_umount_fs(filter->fo_lvfs_ctxt);
1541         filter->fo_sb = 0;
1542         lock_kernel();
1543 err_ops:
1544         fsfilt_put_ops(obd->obd_fsops);
1545         return rc;
1546 }
1547
1548 static int filter_attach(struct obd_device *obd, obd_count len, void *data)
1549 {
1550         struct lprocfs_static_vars lvars;
1551         int rc;
1552
1553         lprocfs_init_vars(filter, &lvars);
1554         rc = lprocfs_obd_attach(obd, lvars.obd_vars);
1555         if (rc != 0)
1556                 return rc;
1557
1558         rc = lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST);
1559         if (rc != 0)
1560                 return rc;
1561
1562         /* Init obdfilter private stats here */
1563         lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES,
1564                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
1565         lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES,
1566                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
1567
1568         return lproc_filter_attach_seqstat(obd);
1569 }
1570
1571 static int filter_detach(struct obd_device *dev)
1572 {
1573         lprocfs_free_obd_stats(dev);
1574         return lprocfs_obd_detach(dev);
1575 }
1576
1577 static int filter_setup(struct obd_device *obd, obd_count len, void *buf)
1578 {
1579         struct lustre_cfg* lcfg = buf;
1580         int rc;
1581         ENTRY;
1582         /* all mount options including errors=remount-ro and asyncdel are passed
1583          * using 4th lcfg param. And it is good, finally we have got rid of
1584          * hardcoded fs types in the code. */
1585         rc = filter_common_setup(obd, len, buf, lcfg->lcfg_inlbuf4);
1586         if (rc)
1587                 RETURN(rc);
1588         rc = filter_post_fs_setup(obd);
1589         RETURN(rc);
1590 }
1591
1592 static int filter_cleanup(struct obd_device *obd, int flags)
1593 {
1594         struct filter_obd *filter = &obd->u.filter;
1595         ll_sbdev_type save_dev;
1596         ENTRY;
1597
1598         if (flags & OBD_OPT_FAILOVER)
1599                 CERROR("%s: shutting down for failover; client state will"
1600                        " be preserved.\n", obd->obd_name);
1601
1602         if (!list_empty(&obd->obd_exports)) {
1603                 CERROR("%s: still has clients!\n", obd->obd_name);
1604                 class_disconnect_exports(obd, flags);
1605                 if (!list_empty(&obd->obd_exports)) {
1606                         CERROR("still has exports after forced cleanup?\n");
1607                         RETURN(-EBUSY);
1608                 }
1609         }
1610
1611         target_cleanup_recovery(obd);
1612
1613         ldlm_namespace_free(obd->obd_namespace, flags & OBD_OPT_FORCE);
1614
1615         if (filter->fo_sb == NULL)
1616                 RETURN(0);
1617
1618         save_dev = ll_sbdev(filter->fo_sb);
1619         filter_post_fs_cleanup(obd);
1620         filter_post(obd);
1621
1622         shrink_dcache_parent(filter->fo_sb->s_root);
1623         filter->fo_sb = 0;
1624
1625         unlock_kernel();
1626         lvfs_umount_fs(filter->fo_lvfs_ctxt);
1627         //destroy_buffers(filter->fo_sb->s_dev);
1628         filter->fo_sb = NULL;
1629         fsfilt_put_ops(obd->obd_fsops);
1630         lock_kernel();
1631
1632         ll_clear_rdonly(save_dev);
1633
1634         RETURN(0);
1635 }
1636
1637 static int filter_connect_post(struct obd_export *exp, unsigned long connect_flags)
1638 {
1639         struct obd_device *obd = exp->exp_obd;
1640         struct filter_export_data *fed;
1641         char str[PTL_NALFMT_SIZE];
1642         struct obd_llogs *llog;
1643         struct llog_ctxt *ctxt;
1644         int rc;
1645         ENTRY;
1646
1647         fed = &exp->exp_filter_data;
1648         if (fed->fed_group < FILTER_MIN_GROUPS)
1649                 RETURN(0);
1650
1651         /* initialize llogs for connections from MDS */
1652         llog = filter_grab_llog_for_group(obd, fed->fed_group, exp);
1653         LASSERT(llog != NULL);
1654
1655         ctxt = llog_get_context(llog, LLOG_UNLINK_REPL_CTXT);
1656         LASSERT(ctxt != NULL);
1657
1658         rc = llog_receptor_accept(ctxt, exp->exp_imp_reverse);
1659         portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number,
1660                         exp->exp_connection->c_peer.peer_id.nid, str);
1661         CDEBUG(D_OTHER, "%s: init llog ctxt for export "LPX64"/%s, group %d\n",
1662                obd->obd_name, exp->exp_connection->c_peer.peer_id.nid,
1663                str, fed->fed_group);
1664
1665         RETURN(rc);
1666 }
1667
1668 /* nearly identical to mds_connect */
1669 static int filter_connect(struct lustre_handle *conn, struct obd_device *obd,
1670                           struct obd_uuid *cluuid,
1671                           struct obd_connect_data *data,
1672                           unsigned long connect_flags)
1673 {
1674         struct obd_export *exp;
1675         struct filter_export_data *fed;
1676         struct filter_client_data *fcd = NULL;
1677         struct filter_obd *filter = &obd->u.filter;
1678         struct lvfs_run_ctxt saved;
1679         int rc, group;
1680         ENTRY;
1681
1682         if (conn == NULL || obd == NULL || cluuid == NULL)
1683                 RETURN(-EINVAL);
1684
1685         rc = class_connect(conn, obd, cluuid);
1686         if (rc)
1687                 RETURN(rc);
1688         exp = class_conn2export(conn);
1689         LASSERT(exp != NULL);
1690
1691         fed = &exp->exp_filter_data;
1692
1693         spin_lock_init(&fed->fed_lock);
1694
1695        /* connection from MDS */
1696         group = connect_flags;
1697         if (obd->obd_replayable) {
1698                 OBD_ALLOC(fcd, sizeof(*fcd));
1699                 if (!fcd) {
1700                         CERROR("filter: out of memory for client data\n");
1701                         GOTO(cleanup, rc = -ENOMEM);
1702                 }
1703
1704                 memcpy(fcd->fcd_uuid, cluuid, sizeof(fcd->fcd_uuid));
1705                 fed->fed_fcd = fcd;
1706                 fed->fed_fcd->fcd_group = group;
1707                 rc = filter_client_add(obd, filter, fed, -1);
1708                 if (rc)
1709                         GOTO(cleanup, rc);
1710         }
1711         CWARN("%s: Received MDS connection ("LPX64"); group %d\n",
1712               obd->obd_name, exp->exp_handle.h_cookie, group);
1713
1714         if (group == 0)
1715                 GOTO(cleanup, rc);
1716
1717         if (fed->fed_group != 0 && fed->fed_group != group) {
1718                 char str[PTL_NALFMT_SIZE];
1719                 portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number,
1720                                 exp->exp_connection->c_peer.peer_id.nid, str);
1721                 CERROR("!!! This export (nid "LPX64"/%s) used object group %d "
1722                        "earlier; now it's trying to use group %d!  This could "
1723                        "be a bug in the MDS.  Tell CFS.\n",
1724                        exp->exp_connection->c_peer.peer_id.nid, str,
1725                        fed->fed_group, group);
1726                 GOTO(cleanup, rc = -EPROTO);
1727         }
1728         fed->fed_group = group;
1729
1730         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1731         rc = filter_read_groups(obd, group, 1);
1732         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1733         if (rc != 0) {
1734                 CERROR("can't read group %u\n", group);
1735                 GOTO(cleanup, rc);
1736         }
1737         rc = filter_group_set_fs_flags(obd, group);
1738         if (rc != 0) {
1739                 CERROR("can't set kml flags %u\n", group);
1740                 GOTO(cleanup, rc);
1741         }
1742
1743 cleanup:
1744         if (rc) {
1745                 if (fcd)
1746                         OBD_FREE(fcd, sizeof(*fcd));
1747                 class_disconnect(exp, 0);
1748         } else {
1749                 class_export_put(exp);
1750         }
1751         return rc;
1752 }
1753
1754 static int filter_precleanup(struct obd_device *obd, int flags)
1755 {
1756         struct filter_group_llog *log;
1757         struct filter_obd *filter;
1758         int rc = 0;
1759         ENTRY;
1760
1761         filter = &obd->u.filter;
1762
1763         spin_lock(&filter->fo_llog_list_lock);
1764         while (!list_empty(&filter->fo_llog_list)) {
1765                 log = list_entry(filter->fo_llog_list.next,
1766                                  struct filter_group_llog, list);
1767                 list_del(&log->list);
1768                 spin_unlock(&filter->fo_llog_list_lock);
1769
1770                 rc = obd_llog_finish(obd, log->llogs, 0);
1771                 if (rc)
1772                         CERROR("failed to cleanup llogging subsystem for %u\n",
1773                                 log->group);
1774                 OBD_FREE(log->llogs, sizeof(*(log->llogs)));
1775                 OBD_FREE(log, sizeof(*log));
1776                 spin_lock(&filter->fo_llog_list_lock);
1777         }
1778         spin_unlock(&filter->fo_llog_list_lock);
1779
1780         rc = obd_llog_finish(obd, &obd->obd_llogs, 0);
1781         if (rc)
1782                 CERROR("failed to cleanup llogging subsystem\n");
1783
1784         RETURN(rc);
1785 }
1786
1787 /* Do extra sanity checks for grant accounting.  We do this at connect,
1788  * disconnect, and statfs RPC time, so it shouldn't be too bad.  We can
1789  * always get rid of it or turn it off when we know accounting is good. */
1790 static void filter_grant_sanity_check(struct obd_device *obd, const char *func)
1791 {
1792         struct filter_export_data *fed;
1793         struct obd_export *exp;
1794         obd_size maxsize = obd->obd_osfs.os_blocks * obd->obd_osfs.os_bsize;
1795         obd_size tot_dirty = 0, tot_pending = 0, tot_granted = 0;
1796         obd_size fo_tot_dirty, fo_tot_pending, fo_tot_granted;
1797         int level = D_CACHE;
1798
1799         if (list_empty(&obd->obd_exports))
1800                 return;
1801
1802         spin_lock(&obd->obd_osfs_lock);
1803         spin_lock(&obd->obd_dev_lock);
1804         list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
1805                 fed = &exp->exp_filter_data;
1806                 if (fed->fed_grant < 0 || fed->fed_pending < 0 ||
1807                     fed->fed_dirty < 0)
1808                         level = D_ERROR;
1809                 if (maxsize > 0) { /* we may not have done a statfs yet */
1810                         LASSERTF(fed->fed_grant + fed->fed_pending <= maxsize,
1811                                  "cli %s/%p %ld+%ld > "LPU64"\n",
1812                                  exp->exp_client_uuid.uuid, exp,
1813                                  fed->fed_grant, fed->fed_pending, maxsize);
1814                         LASSERTF(fed->fed_dirty <= maxsize,
1815                                  "cli %s/%p %ld > "LPU64"\n",
1816                                  exp->exp_client_uuid.uuid, exp,
1817                                  fed->fed_dirty, maxsize);
1818                 }
1819                 CDEBUG(level, "%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
1820                        obd->obd_name, exp->exp_client_uuid.uuid, exp,
1821                        fed->fed_dirty, fed->fed_pending, fed->fed_grant);
1822                 tot_granted += fed->fed_grant + fed->fed_pending;
1823                 tot_pending += fed->fed_pending;
1824                 tot_dirty += fed->fed_dirty;
1825         }
1826         fo_tot_granted = obd->u.filter.fo_tot_granted;
1827         fo_tot_pending = obd->u.filter.fo_tot_pending;
1828         fo_tot_dirty = obd->u.filter.fo_tot_dirty;
1829         spin_unlock(&obd->obd_dev_lock);
1830         spin_unlock(&obd->obd_osfs_lock);
1831
1832         /* Do these assertions outside the spinlocks so we don't kill system */
1833         if (tot_granted != fo_tot_granted)
1834                 CERROR("%s: tot_granted "LPU64" != fo_tot_granted "LPU64"\n",
1835                        func, tot_granted, fo_tot_granted);
1836         if (tot_pending != fo_tot_pending)
1837                 CERROR("%s: tot_pending "LPU64" != fo_tot_pending "LPU64"\n",
1838                        func, tot_pending, fo_tot_pending);
1839         if (tot_dirty != fo_tot_dirty)
1840                 CERROR("%s: tot_dirty "LPU64" != fo_tot_dirty "LPU64"\n",
1841                        func, tot_dirty, fo_tot_dirty);
1842         if (tot_pending > tot_granted)
1843                 CERROR("%s: tot_pending "LPU64" > tot_granted "LPU64"\n",
1844                        func, tot_pending, tot_granted);
1845         if (tot_granted > maxsize)
1846                 CERROR("%s: tot_granted "LPU64" > maxsize "LPU64"\n",
1847                        func, tot_granted, maxsize);
1848         if (tot_dirty > maxsize)
1849                 CERROR("%s: tot_dirty "LPU64" > maxsize "LPU64"\n",
1850                        func, tot_dirty, maxsize);
1851 }
1852
1853 /* Remove this client from the grant accounting totals.  We also remove
1854  * the export from the obd device under the osfs and dev locks to ensure
1855  * that the filter_grant_sanity_check() calculations are always valid.
1856  * The client should do something similar when it invalidates its import. */
1857 static void filter_grant_discard(struct obd_export *exp)
1858 {
1859         struct obd_device *obd = exp->exp_obd;
1860         struct filter_obd *filter = &obd->u.filter;
1861         struct filter_export_data *fed = &exp->exp_filter_data;
1862         int level = D_CACHE;
1863
1864         spin_lock(&obd->obd_osfs_lock);
1865         spin_lock(&exp->exp_obd->obd_dev_lock);
1866         list_del_init(&exp->exp_obd_chain);
1867         spin_unlock(&exp->exp_obd->obd_dev_lock);
1868
1869         if (fed->fed_dirty < 0 || fed->fed_grant < 0 || fed->fed_pending < 0)
1870                 level = D_ERROR;
1871         CDEBUG(level, "%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
1872                obd->obd_name, exp->exp_client_uuid.uuid, exp,
1873                fed->fed_dirty, fed->fed_pending, fed->fed_grant);
1874
1875         LASSERTF(filter->fo_tot_granted >= fed->fed_grant,
1876                  "%s: tot_granted "LPU64" cli %s/%p fed_grant %ld\n",
1877                  obd->obd_name, filter->fo_tot_granted,
1878                  exp->exp_client_uuid.uuid, exp, fed->fed_grant);
1879         filter->fo_tot_granted -= fed->fed_grant;
1880         LASSERTF(filter->fo_tot_pending >= fed->fed_pending,
1881                  "%s: tot_pending "LPU64" cli %s/%p fed_pending %ld\n",
1882                  obd->obd_name, filter->fo_tot_pending,
1883                  exp->exp_client_uuid.uuid, exp, fed->fed_pending);
1884         LASSERTF(filter->fo_tot_dirty >= fed->fed_dirty,
1885                  "%s: tot_dirty "LPU64" cli %s/%p fed_dirty %ld\n",
1886                  obd->obd_name, filter->fo_tot_dirty,
1887                  exp->exp_client_uuid.uuid, exp, fed->fed_dirty);
1888         filter->fo_tot_dirty -= fed->fed_dirty;
1889         fed->fed_dirty = 0;
1890         fed->fed_grant = 0;
1891
1892         spin_unlock(&obd->obd_osfs_lock);
1893 }
1894
1895 static int filter_destroy_export(struct obd_export *exp)
1896 {
1897         ENTRY;
1898
1899         if (exp->exp_filter_data.fed_pending)
1900                 CERROR("%s: cli %s/%p has %lu pending on destroyed export\n",
1901                        exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
1902                        exp, exp->exp_filter_data.fed_pending);
1903
1904         target_destroy_export(exp);
1905
1906         if (exp->exp_obd->obd_replayable)
1907                 filter_client_free(exp, exp->exp_flags);
1908
1909         filter_grant_discard(exp);
1910         if (!(exp->exp_flags & OBD_OPT_FORCE))
1911                 filter_grant_sanity_check(exp->exp_obd, __FUNCTION__);
1912
1913         RETURN(0);
1914 }
1915
1916 static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp)
1917 {
1918         struct filter_group_llog *fglog, *nlog;
1919         struct filter_obd *filter;
1920         int worked = 0, group;
1921         struct llog_ctxt *ctxt;
1922         ENTRY;
1923
1924         filter = &obd->u.filter;
1925
1926         /* we can't sync log holding spinlock. also, we do not want to get
1927          * into livelock. so we do following: loop over MDS's exports in
1928          * group order and skip already synced llogs -bzzz */
1929         do {
1930                 /* look for group with min. number, but > worked */
1931                 fglog = NULL;
1932                 group = 1 << 30;
1933                 spin_lock(&filter->fo_llog_list_lock);
1934                 list_for_each_entry(nlog, &filter->fo_llog_list, list) {
1935
1936                         if (nlog->group <= worked) {
1937                                 /* this group is already synced */
1938                                 continue;
1939                         }
1940
1941                         if (group < nlog->group) {
1942                                 /* we have group with smaller number to sync */
1943                                 continue;
1944                         }
1945
1946                         /* store current minimal group */
1947                         fglog = nlog;
1948                         group = nlog->group;
1949                 }
1950                 spin_unlock(&filter->fo_llog_list_lock);
1951
1952                 if (fglog == NULL)
1953                         break;
1954
1955                 worked = fglog->group;
1956                 if (fglog->exp && (dexp == fglog->exp || dexp == NULL)) {
1957                         ctxt = llog_get_context(fglog->llogs,
1958                                         LLOG_UNLINK_REPL_CTXT);
1959                         LASSERT(ctxt != NULL);
1960                         llog_sync(ctxt, fglog->exp);
1961                 }
1962         } while (fglog != NULL);
1963 }
1964
1965 /* also incredibly similar to mds_disconnect */
1966 static int filter_disconnect(struct obd_export *exp, unsigned long flags)
1967 {
1968         struct obd_device *obd = exp->exp_obd;
1969         unsigned long irqflags;
1970         int rc;
1971         ENTRY;
1972
1973         LASSERT(exp);
1974         class_export_get(exp);
1975
1976         spin_lock_irqsave(&exp->exp_lock, irqflags);
1977         exp->exp_flags = flags;
1978         spin_unlock_irqrestore(&exp->exp_lock, irqflags);
1979
1980         if (!(flags & OBD_OPT_FORCE))
1981                 filter_grant_sanity_check(obd, __FUNCTION__);
1982         filter_grant_discard(exp);
1983
1984         /* Disconnect early so that clients can't keep using export */
1985         rc = class_disconnect(exp, flags);
1986
1987         ldlm_cancel_locks_for_export(exp);
1988
1989         fsfilt_sync(obd, obd->u.filter.fo_sb);
1990
1991         /* flush any remaining cancel messages out to the target */
1992         filter_sync_llogs(obd, exp);
1993         class_export_put(exp);
1994         RETURN(rc);
1995 }
1996
1997 struct dentry *__filter_oa2dentry(struct obd_device *obd,
1998                                   struct obdo *oa, const char *what)
1999 {
2000         struct dentry *dchild = NULL;
2001         obd_gr group = 0;
2002
2003         if (oa->o_valid & OBD_MD_FLGROUP)
2004                 group = oa->o_gr;
2005
2006         dchild = filter_id2dentry(obd, NULL, group, oa->o_id);
2007
2008         if (IS_ERR(dchild)) {
2009                 CERROR("%s error looking up object: "LPU64"\n",
2010                        what, oa->o_id);
2011                 RETURN(dchild);
2012         }
2013
2014         if (dchild->d_inode == NULL) {
2015                 CERROR("%s: %s on non-existent object: "LPU64"\n",
2016                        obd->obd_name, what, oa->o_id);
2017                 f_dput(dchild);
2018                 RETURN(ERR_PTR(-ENOENT));
2019         }
2020
2021         return dchild;
2022 }
2023
2024 static int filter_getattr(struct obd_export *exp, struct obdo *oa,
2025                           struct lov_stripe_md *md)
2026 {
2027         struct dentry *dentry = NULL;
2028         struct obd_device *obd;
2029         int rc = 0;
2030         ENTRY;
2031
2032         obd = class_exp2obd(exp);
2033         if (obd == NULL) {
2034                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
2035                        exp->exp_handle.h_cookie);
2036                 RETURN(-EINVAL);
2037         }
2038
2039         dentry = filter_oa2dentry(obd, oa);
2040         if (IS_ERR(dentry))
2041                 RETURN(PTR_ERR(dentry));
2042
2043         /* Limit the valid bits in the return data to what we actually use */
2044         oa->o_valid = OBD_MD_FLID;
2045         obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS);
2046
2047         f_dput(dentry);
2048         RETURN(rc);
2049 }
2050
2051 /* this is called from filter_truncate() until we have filter_punch() */
2052 static int filter_setattr(struct obd_export *exp, struct obdo *oa,
2053                           struct lov_stripe_md *md, struct obd_trans_info *oti)
2054 {
2055         struct lvfs_run_ctxt saved;
2056         struct filter_obd *filter;
2057         struct dentry *dentry;
2058         struct iattr iattr;
2059         struct ldlm_res_id res_id = { .name = { oa->o_id, 0, oa->o_gr, 0 } };
2060         struct ldlm_resource *res;
2061         void *handle;
2062         int rc, rc2;
2063         ENTRY;
2064
2065         LASSERT(oti != NULL);
2066
2067         dentry = filter_oa2dentry(exp->exp_obd, oa);
2068         if (IS_ERR(dentry))
2069                 RETURN(PTR_ERR(dentry));
2070
2071         filter = &exp->exp_obd->u.filter;
2072
2073         iattr_from_obdo(&iattr, oa, oa->o_valid);
2074
2075         push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
2076         lock_kernel();
2077
2078         if (iattr.ia_valid & ATTR_SIZE)
2079                 down(&dentry->d_inode->i_sem);
2080         handle = fsfilt_start(exp->exp_obd, dentry->d_inode, FSFILT_OP_SETATTR,
2081                               oti);
2082         if (IS_ERR(handle))
2083                 GOTO(out_unlock, rc = PTR_ERR(handle));
2084
2085         /* XXX this could be a rwsem instead, if filter_preprw played along */
2086         if (iattr.ia_valid & ATTR_ATTR_FLAG)
2087                 rc = fsfilt_iocontrol(exp->exp_obd, dentry->d_inode, NULL,
2088                                       EXT3_IOC_SETFLAGS,
2089                                       (long)&iattr.ia_attr_flags);
2090         else
2091                 rc = fsfilt_setattr(exp->exp_obd, dentry, handle, &iattr, 1);
2092         rc = filter_finish_transno(exp, oti, rc);
2093         rc2 = fsfilt_commit(exp->exp_obd, filter->fo_sb, dentry->d_inode,
2094                             handle, exp->exp_sync);
2095         if (rc2) {
2096                 CERROR("error on commit, err = %d\n", rc2);
2097                 if (!rc)
2098                         rc = rc2;
2099         }
2100
2101         res = ldlm_resource_get(exp->exp_obd->obd_namespace, NULL,
2102                                 res_id, LDLM_EXTENT, 0);
2103         if (res != NULL) {
2104                 if (res->lr_namespace->ns_lvbo &&
2105                     res->lr_namespace->ns_lvbo->lvbo_update)
2106                         rc = res->lr_namespace->ns_lvbo->lvbo_update(res, NULL,
2107                                                                      0, 0);
2108                 ldlm_resource_putref(res);
2109         } else if (iattr.ia_valid & ATTR_SIZE) {
2110                 /* called from MDS. */
2111         }
2112
2113         oa->o_valid = OBD_MD_FLID;
2114         obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS);
2115
2116 out_unlock:
2117         if (iattr.ia_valid & ATTR_SIZE)
2118                 up(&dentry->d_inode->i_sem);
2119         unlock_kernel();
2120         pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
2121
2122         f_dput(dentry);
2123         RETURN(rc);
2124 }
2125
2126 /* XXX identical to osc_unpackmd */
2127 static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
2128                            struct lov_mds_md *lmm, int lmm_bytes)
2129 {
2130         int lsm_size;
2131         ENTRY;
2132
2133         if (lmm != NULL) {
2134                 if (lmm_bytes < sizeof (*lmm)) {
2135                         CERROR("lov_mds_md too small: %d, need %d\n",
2136                                lmm_bytes, (int)sizeof(*lmm));
2137                         RETURN(-EINVAL);
2138                 }
2139                 /* XXX LOV_MAGIC etc check? */
2140
2141                 if (lmm->lmm_object_id == cpu_to_le64(0)) {
2142                         CERROR("lov_mds_md: zero lmm_object_id\n");
2143                         RETURN(-EINVAL);
2144                 }
2145         }
2146
2147         lsm_size = lov_stripe_md_size(1);
2148         if (lsmp == NULL)
2149                 RETURN(lsm_size);
2150
2151         if (*lsmp != NULL && lmm == NULL) {
2152                 OBD_FREE(*lsmp, lsm_size);
2153                 *lsmp = NULL;
2154                 RETURN(0);
2155         }
2156
2157         if (*lsmp == NULL) {
2158                 OBD_ALLOC(*lsmp, lsm_size);
2159                 if (*lsmp == NULL)
2160                         RETURN(-ENOMEM);
2161
2162                 loi_init((*lsmp)->lsm_oinfo);
2163         }
2164
2165         if (lmm != NULL) {
2166                 /* XXX zero *lsmp? */
2167                 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
2168                 LASSERT((*lsmp)->lsm_object_id);
2169         }
2170
2171         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
2172
2173         RETURN(lsm_size);
2174 }
2175
2176 static void filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
2177                                       struct filter_obd *filter)
2178 {
2179         struct obdo *doa = NULL;
2180         __u64 last, id;
2181         ENTRY;
2182
2183         LASSERT(oa);
2184         LASSERT(oa->o_gr != 0);
2185         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
2186
2187         doa = obdo_alloc();
2188         if (doa == NULL) {
2189                 CERROR("cannot allocate doa, error %d\n",
2190                        -ENOMEM);
2191                 EXIT;
2192                 return;
2193         }
2194
2195         doa->o_mode = S_IFREG;
2196         doa->o_gr = oa->o_gr;
2197         doa->o_valid = oa->o_valid & (OBD_MD_FLGROUP | OBD_MD_FLID);
2198
2199         set_bit(doa->o_gr, &filter->fo_destroys_in_progress);
2200         down(&filter->fo_create_locks[doa->o_gr]);
2201         if (!test_bit(doa->o_gr, &filter->fo_destroys_in_progress)) {
2202                 CERROR("%s:["LPU64"] destroy_in_progress already cleared\n",
2203                        exp->exp_obd->obd_name, doa->o_gr);
2204                 up(&filter->fo_create_locks[doa->o_gr]);
2205                 GOTO(out_free_doa, 0);
2206         }
2207
2208         last = filter_last_id(filter, doa->o_gr);
2209         CWARN("%s:["LPU64"] deleting orphan objects from "LPU64" to "LPU64"\n",
2210               exp->exp_obd->obd_name, doa->o_gr, oa->o_id + 1, last);
2211         for (id = oa->o_id + 1; id <= last; id++) {
2212                 doa->o_id = id;
2213                 filter_destroy(exp, doa, NULL, NULL);
2214         }
2215
2216         CDEBUG(D_HA, "%s:["LPU64"] after destroy: set last_objids = "LPU64"\n",
2217                exp->exp_obd->obd_name, doa->o_gr, oa->o_id);
2218
2219         filter_set_last_id(filter, doa->o_gr, oa->o_id);
2220
2221         clear_bit(doa->o_gr, &filter->fo_destroys_in_progress);
2222         up(&filter->fo_create_locks[doa->o_gr]);
2223
2224         EXIT;
2225 out_free_doa:
2226         obdo_free(doa);
2227 }
2228
2229 /* returns a negative error or a nonnegative number of files to create */
2230 static int filter_should_precreate(struct obd_export *exp, struct obdo *oa,
2231                                    obd_gr group)
2232 {
2233         struct obd_device *obd = exp->exp_obd;
2234         struct filter_obd *filter = &obd->u.filter;
2235         int diff, rc;
2236         ENTRY;
2237
2238         diff = oa->o_id - filter_last_id(filter, oa->o_gr);
2239         CDEBUG(D_INFO, "filter_last_id() = "LPU64" -> diff = %d\n",
2240                filter_last_id(filter, oa->o_gr), diff);
2241
2242         /* delete orphans request */
2243         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
2244             (oa->o_flags & OBD_FL_DELORPHAN)) {
2245                 if (diff >= 0)
2246                         RETURN(diff);
2247                 if (-diff > OST_MAX_PRECREATE) {
2248                         CERROR("ignoring bogus orphan destroy request: obdid "
2249                                LPU64" last_id "LPU64"\n",
2250                                oa->o_id, filter_last_id(filter, oa->o_gr));
2251                         RETURN(-EINVAL);
2252                 }
2253                 filter_destroy_precreated(exp, oa, filter);
2254                 rc = filter_update_last_objid(obd, group, 0);
2255                 if (rc)
2256                         CERROR("unable to write lastobjid, but orphans"
2257                                "were deleted\n");
2258                 RETURN(0);
2259         } else {
2260                 /* only precreate if group == 0 and o_id is specfied */
2261                 if (!(oa->o_valid & OBD_FL_DELORPHAN) &&
2262                     (/*group != 0 ||*/ oa->o_id == 0))
2263                         RETURN(1);
2264
2265                 LASSERTF(diff >= 0, LPU64" - "LPU64" = %d\n", oa->o_id,
2266                          filter_last_id(filter, oa->o_gr), diff);
2267                 RETURN(diff);
2268         }
2269 }
2270 static int filter_precreate_rec(struct obd_device *obd, struct dentry *dentry,
2271                                 int *number, struct obdo *oa)
2272 {
2273         int rc;
2274         ENTRY;
2275
2276         rc = fsfilt_precreate_rec(obd, dentry, number, oa);
2277
2278         RETURN(rc);
2279 }
2280
2281 static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs,
2282                          unsigned long max_age)
2283 {
2284         struct filter_obd *filter = &obd->u.filter;
2285         int blockbits = filter->fo_sb->s_blocksize_bits;
2286         int rc;
2287         ENTRY;
2288
2289         /* at least try to account for cached pages.  its still racey and
2290          * might be under-reporting if clients haven't announced their
2291          * caches with brw recently */
2292         spin_lock(&obd->obd_osfs_lock);
2293         rc = fsfilt_statfs(obd, filter->fo_sb, max_age);
2294         memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
2295         spin_unlock(&obd->obd_osfs_lock);
2296
2297         CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64
2298                " pending "LPU64" free "LPU64" avail "LPU64"\n",
2299                filter->fo_tot_dirty, filter->fo_tot_granted,
2300                filter->fo_tot_pending,
2301                osfs->os_bfree << blockbits, osfs->os_bavail << blockbits);
2302
2303         filter_grant_sanity_check(obd, __FUNCTION__);
2304
2305         osfs->os_bavail -= min(osfs->os_bavail,
2306                                (filter->fo_tot_dirty + filter->fo_tot_pending +
2307                                 osfs->os_bsize -1) >> blockbits);
2308
2309         RETURN(rc);
2310 }
2311
2312 /* We rely on the fact that only one thread will be creating files in a given
2313  * group at a time, which is why we don't need an atomic filter_get_new_id.
2314  * Even if we had that atomic function, the following race would exist:
2315  *
2316  * thread 1: gets id x from filter_next_id
2317  * thread 2: gets id (x + 1) from filter_next_id
2318  * thread 2: creates object (x + 1)
2319  * thread 1: tries to create object x, gets -ENOSPC
2320  */
2321 static int filter_precreate(struct obd_device *obd, struct obdo *oa,
2322                             obd_gr group, int *num)
2323 {
2324         struct dentry *dchild = NULL, *dparent = NULL;
2325         int err = 0, rc = 0, recreate_obj = 0, i;
2326         struct filter_obd *filter;
2327         void *handle = NULL;
2328         void *lock = NULL;
2329         struct obd_statfs *osfs;
2330         unsigned long enough_time = jiffies + (obd_timeout * HZ) / 3;
2331         __u64 next_id;
2332         ENTRY;
2333
2334         filter = &obd->u.filter;
2335
2336         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
2337             (oa->o_flags & OBD_FL_RECREATE_OBJS)) {
2338                 recreate_obj = 1;
2339         } else {
2340                 OBD_ALLOC(osfs, sizeof(*osfs));
2341                 if (osfs == NULL)
2342                         RETURN(-ENOMEM);
2343                 rc = filter_statfs(obd, osfs, jiffies-HZ);
2344                 if (rc == 0 && osfs->os_bavail < (osfs->os_blocks >> 10)) {
2345                         CDEBUG(D_HA, "OST out of space! avail "LPU64"\n",
2346                               osfs->os_bavail<<filter->fo_sb->s_blocksize_bits);
2347                         *num = 0;
2348                         rc = -ENOSPC;
2349                 }
2350                 OBD_FREE(osfs, sizeof(*osfs));
2351                 if (rc) {
2352                         RETURN(rc);
2353                 }
2354         }
2355
2356         CDEBUG(D_HA, "%s: precreating %d objects\n", obd->obd_name, *num);
2357
2358         down(&filter->fo_create_locks[group]);
2359
2360         for (i = 0; i < *num && err == 0; i++) {
2361                 int cleanup_phase = 0;
2362
2363                 if (test_bit(group, &filter->fo_destroys_in_progress)) {
2364                         CWARN("%s: precreate aborted by destroy\n",
2365                               obd->obd_name);
2366                         break;
2367                 }
2368
2369                 if (recreate_obj) {
2370                         __u64 last_id;
2371                         next_id = oa->o_id;
2372                         last_id = filter_last_id(filter, group);
2373                         if (next_id > last_id) {
2374                                 CERROR("Error: Trying to recreate obj greater"
2375                                        "than last id "LPD64" > "LPD64"\n",
2376                                        next_id, last_id);
2377                                 GOTO(cleanup, rc = -EINVAL);
2378                         }
2379                 } else {
2380                         next_id = filter_last_id(filter, group) + 1;
2381                 }
2382
2383                 CDEBUG(D_INFO, "precreate objid "LPU64"\n", next_id);
2384
2385                 dparent = filter_parent_lock(obd, group, next_id, &lock);
2386                 if (IS_ERR(dparent))
2387                         GOTO(cleanup, rc = PTR_ERR(dparent));
2388                 cleanup_phase = 1;
2389
2390                 /*only do precreate rec record. so clean kml flags here*/
2391                 fsfilt_clear_fs_flags(obd, dparent->d_inode, SM_DO_REC);
2392
2393                 dchild = filter_id2dentry(obd, dparent, group, next_id);
2394                 if (IS_ERR(dchild))
2395                         GOTO(cleanup, rc = PTR_ERR(dchild));
2396                 cleanup_phase = 2;
2397
2398                 if (dchild->d_inode != NULL) {
2399                         /* This would only happen if lastobjid was bad on disk*/
2400                         /* Could also happen if recreating missing obj but
2401                          * already exists
2402                          */
2403                         if (recreate_obj) {
2404                                 CERROR("%s: recreating existing object %.*s?\n",
2405                                        obd->obd_name, dchild->d_name.len,
2406                                        dchild->d_name.name);
2407                         } else {
2408                                 CERROR("%s: Serious error: objid %.*s already "
2409                                        "exists; is this filesystem corrupt?\n",
2410                                        obd->obd_name, dchild->d_name.len,
2411                                        dchild->d_name.name);
2412                                 LBUG();
2413                         }
2414                         GOTO(cleanup, rc = -EEXIST);
2415                 }
2416
2417                 handle = fsfilt_start_log(obd, dparent->d_inode,
2418                                           FSFILT_OP_CREATE, NULL, 1);
2419                 if (IS_ERR(handle))
2420                         GOTO(cleanup, rc = PTR_ERR(handle));
2421                 cleanup_phase = 3;
2422
2423                 rc = ll_vfs_create(dparent->d_inode, dchild, S_IFREG, NULL);
2424                 if (rc) {
2425                         CERROR("create failed rc = %d\n", rc);
2426                         GOTO(cleanup, rc);
2427                 }
2428
2429                 if (!recreate_obj) {
2430                         filter_set_last_id(filter, group, next_id);
2431                         err = filter_update_last_objid(obd, group, 0);
2432                         if (err)
2433                                 CERROR("unable to write lastobjid "
2434                                        "but file created\n");
2435                 }
2436                 fsfilt_set_fs_flags(obd, dparent->d_inode, SM_DO_REC);
2437
2438         cleanup:
2439                 switch(cleanup_phase) {
2440                 case 3:
2441                         err = fsfilt_commit(obd, filter->fo_sb,
2442                                             dparent->d_inode, handle, 0);
2443                         if (err) {
2444                                 CERROR("error on commit, err = %d\n", err);
2445                                 if (!rc)
2446                                         rc = err;
2447                         }
2448                 case 2:
2449                         f_dput(dchild);
2450                 case 1:
2451                         filter_parent_unlock(dparent, lock);
2452                 case 0:
2453                         break;
2454                 }
2455
2456                 if (rc)
2457                         break;
2458                 if (time_after(jiffies, enough_time)) {
2459                         CDEBUG(D_INODE,"%s: precreate slow - want %d got %d \n",
2460                                obd->obd_name, *num, i);
2461                         break;
2462                 }
2463         }
2464
2465         *num = i;
2466
2467         /* check if we have an error after ll_vfs_create(). It is possible that
2468          * there will be say -ENOSPC and we will leak it. */
2469         if (rc == 0)
2470                 rc = filter_precreate_rec(obd, dparent, num, oa);
2471
2472         up(&filter->fo_create_locks[group]);
2473
2474         CDEBUG(D_HA, "%s: server last_objid for group "LPU64": "LPU64"\n",
2475                obd->obd_name, group, filter->fo_last_objids[group]);
2476
2477         CDEBUG(D_HA, "%s: filter_precreate() created %d objects\n",
2478                obd->obd_name, i);
2479
2480         RETURN(rc);
2481 }
2482
2483 static int filter_create(struct obd_export *exp, struct obdo *oa,
2484                          struct lov_stripe_md **ea, struct obd_trans_info *oti)
2485 {
2486         struct obd_device *obd = NULL;
2487         struct filter_obd *filter;
2488         struct lvfs_run_ctxt saved;
2489         struct lov_stripe_md *lsm = NULL;
2490         struct filter_export_data *fed;
2491         char str[PTL_NALFMT_SIZE];
2492         int group = oa->o_gr, rc = 0, diff, recreate_objs = 0;
2493         ENTRY;
2494
2495         if (!(oa->o_valid & OBD_MD_FLGROUP) || group == 0) {
2496                 portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number,
2497                                 exp->exp_connection->c_peer.peer_id.nid, str);
2498                 CERROR("!!! nid "LPX64"/%s sent invalid object group %d\n",
2499                        exp->exp_connection->c_peer.peer_id.nid, str, group);
2500                 RETURN(-EINVAL);
2501         }
2502
2503         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
2504             (oa->o_flags & OBD_FL_RECREATE_OBJS))
2505                 recreate_objs = 1;
2506
2507         obd = exp->exp_obd;
2508         fed = &exp->exp_filter_data;
2509         filter = &obd->u.filter;
2510
2511         if (fed->fed_group != group && !recreate_objs &&
2512             !(oa->o_valid & OBD_MD_REINT)) {
2513                 portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number,
2514                                 exp->exp_connection->c_peer.peer_id.nid, str);
2515                 CERROR("!!! This export (nid "LPX64"/%s) used object group %d "
2516                        "earlier; now it's trying to use group %d!  This could "
2517                        "be a bug in the MDS.  Tell CFS.\n",
2518                        exp->exp_connection->c_peer.peer_id.nid, str,
2519                        fed->fed_group, group);
2520                 RETURN(-ENOTUNIQ);
2521         }
2522
2523         CDEBUG(D_INFO, "filter_create(od->o_gr=%d,od->o_id="LPU64")\n",
2524                group, oa->o_id);
2525         if (ea != NULL) {
2526                 lsm = *ea;
2527                 if (lsm == NULL) {
2528                         rc = obd_alloc_memmd(exp, &lsm);
2529                         if (rc < 0)
2530                                 RETURN(rc);
2531                 }
2532         }
2533
2534         obd = exp->exp_obd;
2535         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2536
2537         if (oa->o_valid & OBD_MD_REINT) {
2538                 int num = *((int*)oa->o_inline);
2539                 rc = filter_precreate(obd, oa, oa->o_gr, &num);
2540         } else if (recreate_objs) {
2541                 if (oa->o_id > filter_last_id(&obd->u.filter, group)) {
2542                         CERROR("recreate objid "LPU64" > last id "LPU64"\n",
2543                                oa->o_id, filter_last_id(&obd->u.filter, group));
2544                         rc = -EINVAL;
2545                 } else {
2546                         diff = 1;
2547                         rc = filter_precreate(obd, oa, group, &diff);
2548                 }
2549         } else {
2550                 diff = filter_should_precreate(exp, oa, group);
2551                 if (diff > 0) {
2552                         oa->o_id = filter_last_id(&obd->u.filter, group);
2553                         rc = filter_precreate(obd, oa, group, &diff);
2554                         oa->o_id = filter_last_id(&obd->u.filter, oa->o_gr);
2555                         oa->o_valid = OBD_MD_FLID;
2556                 }
2557         }
2558
2559         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2560         if (rc && ea != NULL && *ea != lsm) {
2561                 obd_free_memmd(exp, &lsm);
2562         } else if (rc == 0 && ea != NULL) {
2563                 /* XXX LOV STACKING: the lsm that is passed to us from
2564                  * LOV does not have valid lsm_oinfo data structs, so
2565                  * don't go touching that.  This needs to be fixed in a
2566                  * big way. */
2567                 lsm->lsm_object_id = oa->o_id;
2568                 lsm->lsm_object_gr = oa->o_gr;
2569                 *ea = lsm;
2570         }
2571
2572         RETURN(rc);
2573 }
2574
2575 static int filter_destroy(struct obd_export *exp, struct obdo *oa,
2576                           struct lov_stripe_md *ea, struct obd_trans_info *oti)
2577 {
2578         int rc, rc2, cleanup_phase = 0, have_prepared = 0;
2579         struct dentry *dchild = NULL, *dparent = NULL;
2580         struct llog_cookie *fcc = NULL;
2581         struct lvfs_run_ctxt saved;
2582         struct filter_obd *filter;
2583         struct obd_device *obd;
2584         void *handle = NULL;
2585         void *lock = NULL;
2586         struct iattr iattr;
2587         ENTRY;
2588
2589         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
2590
2591         obd = exp->exp_obd;
2592         filter = &obd->u.filter;
2593
2594         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2595
2596  acquire_locks:
2597         dparent = filter_parent_lock(obd, oa->o_gr, oa->o_id, &lock);
2598         if (IS_ERR(dparent))
2599                 GOTO(cleanup, rc = PTR_ERR(dparent));
2600         cleanup_phase = 1;
2601
2602         dchild = filter_id2dentry(obd, dparent, oa->o_gr, oa->o_id);
2603         if (IS_ERR(dchild))
2604                 GOTO(cleanup, rc = PTR_ERR(dchild));
2605         cleanup_phase = 2;
2606
2607         if (dchild->d_inode == NULL) {
2608                 CDEBUG(D_INODE, "destroying non-existent object "LPU64"\n",
2609                        oa->o_id);
2610                 GOTO(cleanup, rc = -ENOENT);
2611         }
2612
2613         if (!have_prepared) {
2614                 /* If we're really going to destroy the object, get ready
2615                  * by getting the clients to discard their cached data.
2616                  *
2617                  * We have to drop the parent lock, because
2618                  * filter_prepare_destroy will acquire a PW on the object, and
2619                  * we don't want to deadlock with an incoming write to the
2620                  * object, which has the extent PW and then wants to get the
2621                  * parent dentry to do the lookup.
2622                  *
2623                  * We dput the child because it's not worth the extra
2624                  * complication of condition the above code to skip it on the
2625                  * second time through. */
2626                 f_dput(dchild);
2627                 filter_parent_unlock(dparent, lock);
2628
2629                 filter_prepare_destroy(obd, oa->o_id, oa->o_gr);
2630                 have_prepared = 1;
2631                 goto acquire_locks;
2632         }
2633
2634         /* Our MDC connection is established by the MDS to us */
2635         if (oa->o_valid & OBD_MD_FLCOOKIE) {
2636                 OBD_ALLOC(fcc, sizeof(*fcc));
2637                 if (fcc != NULL)
2638                         memcpy(fcc, obdo_logcookie(oa), sizeof(*fcc));
2639         }
2640
2641         /* we're gonna truncate it first in order to avoid possible
2642          * deadlock:
2643          *      P1                      P2
2644          * open trasaction      open transaction
2645          * down(i_zombie)       down(i_zombie)
2646          *                      restart transaction
2647          * (see BUG 4180) -bzzz
2648          */
2649         down(&dchild->d_inode->i_sem);
2650         handle = fsfilt_start_log(obd, dparent->d_inode,FSFILT_OP_SETATTR,NULL,1);
2651         if (IS_ERR(handle)) {
2652                 up(&dchild->d_inode->i_sem);
2653                 GOTO(cleanup, rc = PTR_ERR(handle));
2654         }
2655
2656         iattr.ia_valid = ATTR_SIZE;
2657         iattr.ia_size = 0;
2658         rc = fsfilt_setattr(obd, dchild, handle, &iattr, 1);
2659
2660         rc2 = fsfilt_commit(obd, filter->fo_sb, dparent->d_inode, handle, 0);
2661         up(&dchild->d_inode->i_sem);
2662         if (rc)
2663                 GOTO(cleanup, rc);
2664         if (rc2)
2665                 GOTO(cleanup, rc = rc2);
2666
2667         handle = fsfilt_start_log(obd, dparent->d_inode,FSFILT_OP_UNLINK,oti,1);
2668         if (IS_ERR(handle))
2669                 GOTO(cleanup, rc = PTR_ERR(handle));
2670
2671         cleanup_phase = 3;
2672
2673         rc = filter_destroy_internal(obd, oa->o_id, dparent, dchild);
2674
2675 cleanup:
2676         switch(cleanup_phase) {
2677         case 3:
2678                 if (fcc != NULL) {
2679                         if (oti != NULL)
2680                                 fsfilt_add_journal_cb(obd, filter->fo_sb, 0,
2681                                                       oti->oti_handle,
2682                                                       filter_cancel_cookies_cb,
2683                                                       fcc);
2684                         else
2685                                 fsfilt_add_journal_cb(obd, filter->fo_sb, 0,
2686                                                       handle,
2687                                                       filter_cancel_cookies_cb,
2688                                                       fcc);
2689                 }
2690                 rc = filter_finish_transno(exp, oti, rc);
2691                 rc2 = fsfilt_commit(obd, filter->fo_sb, dparent->d_inode,
2692                                     handle, exp->exp_sync);
2693                 if (rc2) {
2694                         CERROR("error on commit, err = %d\n", rc2);
2695                         if (!rc)
2696                                 rc = rc2;
2697                 }
2698         case 2:
2699                 f_dput(dchild);
2700         case 1:
2701                 filter_parent_unlock(dparent, lock);
2702         case 0:
2703                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2704                 break;
2705         default:
2706                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2707                 LBUG();
2708         }
2709
2710         RETURN(rc);
2711 }
2712
2713 /* NB start and end are used for punch, but not truncate */
2714 static int filter_truncate(struct obd_export *exp, struct obdo *oa,
2715                            struct lov_stripe_md *lsm,
2716                            obd_off start, obd_off end,
2717                            struct obd_trans_info *oti)
2718 {
2719         int error;
2720         ENTRY;
2721
2722         if (end != OBD_OBJECT_EOF)
2723                 CERROR("PUNCH not supported, only truncate: end = "LPX64"\n",
2724                        end);
2725
2726         CDEBUG(D_INODE, "calling truncate for object "LPU64", valid = "LPU64", "
2727                "o_size = "LPD64"\n", oa->o_id, oa->o_valid, start);
2728         oa->o_size = start;
2729         error = filter_setattr(exp, oa, NULL, oti);
2730         RETURN(error);
2731 }
2732
2733 static int filter_sync(struct obd_export *exp, struct obdo *oa,
2734                        struct lov_stripe_md *lsm, obd_off start, obd_off end)
2735 {
2736         struct obd_device *obd = exp->exp_obd;
2737         struct lvfs_run_ctxt saved;
2738         struct filter_obd *filter;
2739         struct dentry *dentry;
2740         int rc, rc2;
2741         ENTRY;
2742
2743         filter = &obd->u.filter;
2744
2745         /* an objid of zero is taken to mean "sync whole filesystem" */
2746         if (!oa || !(oa->o_valid & OBD_MD_FLID)) {
2747                 rc = fsfilt_sync(obd, filter->fo_sb);
2748                 /* flush any remaining cancel messages out to the target */
2749                 filter_sync_llogs(obd, NULL);
2750                 RETURN(rc);
2751         }
2752
2753         dentry = filter_oa2dentry(obd, oa);
2754         if (IS_ERR(dentry))
2755                 RETURN(PTR_ERR(dentry));
2756
2757         push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
2758
2759         down(&dentry->d_inode->i_sem);
2760         rc = filemap_fdatasync(dentry->d_inode->i_mapping);
2761         if (rc == 0) {
2762                 /* just any file to grab fsync method - "file" arg unused */
2763                 struct file *file = filter->fo_rcvd_filp;
2764
2765                 if (file->f_op && file->f_op->fsync)
2766                         rc = file->f_op->fsync(NULL, dentry, 1);
2767
2768                 rc2 = filemap_fdatawait(dentry->d_inode->i_mapping);
2769                 if (!rc)
2770                         rc = rc2;
2771         }
2772         up(&dentry->d_inode->i_sem);
2773
2774         oa->o_valid = OBD_MD_FLID;
2775         obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS);
2776
2777         pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
2778
2779         f_dput(dentry);
2780         RETURN(rc);
2781 }
2782
2783 static int filter_get_info(struct obd_export *exp, __u32 keylen,
2784                            void *key, __u32 *vallen, void *val)
2785 {
2786         struct filter_export_data *fed = &exp->exp_filter_data;
2787         struct obd_device *obd;
2788         ENTRY;
2789
2790         obd = class_exp2obd(exp);
2791         if (obd == NULL) {
2792                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
2793                        exp->exp_handle.h_cookie);
2794                 RETURN(-EINVAL);
2795         }
2796
2797         if (keylen == strlen("blocksize") &&
2798             memcmp(key, "blocksize", keylen) == 0) {
2799                 __u32 *blocksize = val;
2800                 *vallen = sizeof(*blocksize);
2801                 *blocksize = obd->u.filter.fo_sb->s_blocksize;
2802                 RETURN(0);
2803         }
2804
2805         if (keylen == strlen("blocksize_bits") &&
2806             memcmp(key, "blocksize_bits", keylen) == 0) {
2807                 __u32 *blocksize_bits = val;
2808                 *vallen = sizeof(*blocksize_bits);
2809                 *blocksize_bits = obd->u.filter.fo_sb->s_blocksize_bits;
2810                 RETURN(0);
2811         }
2812
2813         if (keylen >= strlen("last_id") && memcmp(key, "last_id", 7) == 0) {
2814                 obd_id *last_id = val;
2815                 *last_id = filter_last_id(&obd->u.filter, fed->fed_group);
2816                 RETURN(0);
2817         }
2818         if (keylen >= strlen("reint_log") && memcmp(key, "reint_log", 9) == 0) {
2819                 /*Get log_context handle*/
2820                 unsigned long *llh_handle = val;
2821                 *vallen = sizeof(unsigned long);
2822                 *llh_handle = (unsigned long)obd->obd_llog_ctxt[LLOG_REINT_ORIG_CTXT];
2823                 RETURN(0);
2824         }
2825         if (keylen >= strlen("cache_sb") && memcmp(key, "cache_sb", 8) == 0) {
2826                 /*Get log_context handle*/
2827                 unsigned long *sb = val;
2828                 *vallen = sizeof(unsigned long);
2829                 *sb = (unsigned long)obd->u.filter.fo_sb;
2830                 RETURN(0);
2831         }
2832
2833         CDEBUG(D_IOCTL, "invalid key\n");
2834         RETURN(-EINVAL);
2835 }
2836
2837 struct obd_llogs *filter_grab_llog_for_group(struct obd_device *obd, int group,
2838                                              struct obd_export *export)
2839 {
2840         struct filter_group_llog *fglog, *nlog;
2841         char name[32] = "CATLIST";
2842         struct filter_obd *filter;
2843         struct llog_ctxt *ctxt;
2844         struct list_head *cur;
2845         int rc;
2846
2847         filter = &obd->u.filter;
2848
2849         spin_lock(&filter->fo_llog_list_lock);
2850         list_for_each(cur, &filter->fo_llog_list) {
2851                 fglog = list_entry(cur, struct filter_group_llog, list);
2852                 if (fglog->group == group) {
2853                         if (!(fglog->exp == NULL || fglog->exp == export || export == NULL))
2854                                 CWARN("%s: export for group %d changes: 0x%p -> 0x%p\n",
2855                                       obd->obd_name, group, fglog->exp, export);
2856                         spin_unlock(&filter->fo_llog_list_lock);
2857                         goto init;
2858                 }
2859         }
2860         spin_unlock(&filter->fo_llog_list_lock);
2861
2862         if (export == NULL)
2863                 RETURN(NULL);
2864
2865         OBD_ALLOC(fglog, sizeof(*fglog));
2866         if (fglog == NULL)
2867                 RETURN(NULL);
2868         fglog->group = group;
2869
2870         OBD_ALLOC(fglog->llogs, sizeof(struct obd_llogs));
2871         if (fglog->llogs == NULL) {
2872                 OBD_FREE(fglog, sizeof(*fglog));
2873                 RETURN(NULL);
2874         }
2875
2876         spin_lock(&filter->fo_llog_list_lock);
2877         list_for_each(cur, &filter->fo_llog_list) {
2878                 nlog = list_entry(cur, struct filter_group_llog, list);
2879                 LASSERT(nlog->group != group);
2880         }
2881         list_add(&fglog->list, &filter->fo_llog_list);
2882         spin_unlock(&filter->fo_llog_list_lock);
2883
2884         rc = obd_llog_cat_initialize(obd, fglog->llogs, 1, name);
2885         if (rc) {
2886                 OBD_FREE(fglog->llogs, sizeof(*(fglog->llogs)));
2887                 OBD_FREE(fglog, sizeof(*fglog));
2888                 RETURN(NULL);
2889         }
2890
2891 init:
2892         if (export) {
2893                 fglog->exp = export;
2894                 ctxt = llog_get_context(fglog->llogs, LLOG_UNLINK_REPL_CTXT);
2895                 LASSERT(ctxt != NULL);
2896
2897                 llog_receptor_accept(ctxt, export->exp_imp_reverse);
2898         }
2899         CDEBUG(D_OTHER, "%s: new llog 0x%p for group %u\n",
2900                obd->obd_name, fglog->llogs, group);
2901
2902         RETURN(fglog->llogs);
2903 }
2904
2905 int filter_iocontrol(unsigned int cmd, struct obd_export *exp,
2906                      int len, void *karg, void *uarg)
2907 {
2908         struct obd_device *obd = exp->exp_obd;
2909         struct obd_ioctl_data *data = karg;
2910         int rc = 0;
2911
2912         switch (cmd) {
2913         case OBD_IOC_ABORT_RECOVERY:
2914                 target_stop_recovery_thread(obd);
2915                 RETURN(0);
2916
2917         case OBD_IOC_SET_READONLY: {
2918                 void *handle;
2919                 struct super_block *sb = obd->u.filter.fo_sb;
2920                 struct inode *inode = sb->s_root->d_inode;
2921                 BDEVNAME_DECLARE_STORAGE(tmp);
2922                 CERROR("setting device %s read-only\n",
2923                        ll_bdevname(sb, tmp));
2924
2925                 handle = fsfilt_start(obd, inode, FSFILT_OP_MKNOD, NULL);
2926                 LASSERT(handle);
2927                 (void)fsfilt_commit(obd, sb, inode, handle, 1);
2928
2929                 ll_set_rdonly(ll_sbdev(obd->u.filter.fo_sb));
2930                 RETURN(0);
2931         }
2932
2933         case OBD_IOC_CATLOGLIST: {
2934                 rc = llog_catalog_list(obd, 1, data);
2935                 RETURN(rc);
2936         }
2937         case OBD_IOC_LLOG_CANCEL:
2938         case OBD_IOC_LLOG_REMOVE:
2939         case OBD_IOC_LLOG_INFO:
2940         case OBD_IOC_LLOG_PRINT: {
2941                 /* FIXME to be finished */
2942                 RETURN(-EOPNOTSUPP);
2943 /*
2944                 struct llog_ctxt *ctxt = NULL;
2945
2946                 push_ctxt(&saved, &ctxt->loc_ctxt, NULL);
2947                 rc = llog_ioctl(ctxt, cmd, data);
2948                 pop_ctxt(&saved, &ctxt->loc_ctxt, NULL);
2949
2950                 RETURN(rc);
2951 */
2952         }
2953
2954
2955         default:
2956                 RETURN(-EINVAL);
2957         }
2958         RETURN(0);
2959 }
2960
2961 static struct llog_operations filter_unlink_repl_logops;
2962 static struct llog_operations filter_size_orig_logops = {
2963         lop_setup: llog_obd_origin_setup,
2964         lop_cleanup: llog_catalog_cleanup,
2965         lop_add: llog_catalog_add,
2966 };
2967
2968 static int filter_llog_init(struct obd_device *obd, struct obd_llogs *llogs,
2969                             struct obd_device *tgt, int count,
2970                             struct llog_catid *catid)
2971 {
2972         struct llog_ctxt *ctxt;
2973         int rc;
2974         ENTRY;
2975
2976         filter_unlink_repl_logops = llog_client_ops;
2977         filter_unlink_repl_logops.lop_cancel = llog_obd_repl_cancel;
2978         filter_unlink_repl_logops.lop_connect = llog_repl_connect;
2979         filter_unlink_repl_logops.lop_sync = llog_obd_repl_sync;
2980
2981         rc = obd_llog_setup(obd, llogs, LLOG_UNLINK_REPL_CTXT, tgt, 0, NULL,
2982                         &filter_unlink_repl_logops);
2983         if (rc)
2984                 RETURN(rc);
2985         /* FIXME - assign unlink_cb for filter's recovery */
2986         ctxt = llog_get_context(llogs, LLOG_UNLINK_REPL_CTXT);
2987         ctxt->llog_proc_cb = filter_recov_log_unlink_cb;
2988
2989         /* FIXME - count should be 1 to setup size log */
2990         rc = obd_llog_setup(obd, llogs, LLOG_SIZE_ORIG_CTXT, tgt, 0,
2991                             &catid->lci_logid, &filter_size_orig_logops);
2992         RETURN(rc);
2993 }
2994
2995 static int filter_llog_finish(struct obd_device *obd,
2996                               struct obd_llogs *llogs, int count)
2997 {
2998         int rc;
2999         ENTRY;
3000
3001         rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_UNLINK_REPL_CTXT));
3002         if (rc)
3003                 RETURN(rc);
3004
3005         rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_SIZE_ORIG_CTXT));
3006         RETURN(rc);
3007 }
3008
3009 static int filter_llog_connect(struct obd_export *exp,
3010                                struct llogd_conn_body *body)
3011 {
3012         struct obd_device *obd = exp->exp_obd;
3013         struct llog_ctxt *ctxt;
3014         struct obd_llogs *llog;
3015         int rc;
3016         ENTRY;
3017
3018         CDEBUG(D_OTHER, "handle connect for %s: %u/%u/%u\n", obd->obd_name,
3019                (unsigned) body->lgdc_logid.lgl_ogr,
3020                (unsigned) body->lgdc_logid.lgl_oid,
3021                (unsigned) body->lgdc_logid.lgl_ogen);
3022         llog = filter_grab_llog_for_group(obd, body->lgdc_logid.lgl_ogr, exp);
3023         LASSERT(llog != NULL);
3024         ctxt = llog_get_context(llog, body->lgdc_ctxt_idx);
3025         rc = llog_connect(ctxt, 1, &body->lgdc_logid,
3026                           &body->lgdc_gen, NULL);
3027         if (rc != 0)
3028                 CERROR("failed to connect\n");
3029
3030         RETURN(rc);
3031 }
3032
3033 static struct dentry *filter_lvfs_id2dentry(__u64 id, __u32 gen,
3034                                             __u64 gr, void *data)
3035 {
3036         return filter_id2dentry(data, NULL, gr, id);
3037 }
3038
3039 static struct lvfs_callback_ops filter_lvfs_ops = {
3040         l_id2dentry:     filter_lvfs_id2dentry,
3041 };
3042
3043 static struct obd_ops filter_obd_ops = {
3044         .o_owner          = THIS_MODULE,
3045         .o_attach         = filter_attach,
3046         .o_detach         = filter_detach,
3047         .o_get_info       = filter_get_info,
3048         .o_setup          = filter_setup,
3049         .o_precleanup     = filter_precleanup,
3050         .o_cleanup        = filter_cleanup,
3051         .o_connect        = filter_connect,
3052         .o_connect_post   = filter_connect_post,
3053         .o_disconnect     = filter_disconnect,
3054         .o_statfs         = filter_statfs,
3055         .o_getattr        = filter_getattr,
3056         .o_unpackmd       = filter_unpackmd,
3057         .o_create         = filter_create,
3058         .o_setattr        = filter_setattr,
3059         .o_destroy        = filter_destroy,
3060         .o_brw            = filter_brw,
3061         .o_punch          = filter_truncate,
3062         .o_sync           = filter_sync,
3063         .o_preprw         = filter_preprw,
3064         .o_commitrw       = filter_commitrw,
3065         .o_do_cow         = filter_do_cow,
3066         .o_write_extents  = filter_write_extents,
3067         .o_destroy_export = filter_destroy_export,
3068         .o_llog_init      = filter_llog_init,
3069         .o_llog_finish    = filter_llog_finish,
3070         .o_llog_connect   = filter_llog_connect,
3071         .o_iocontrol      = filter_iocontrol,
3072 };
3073
3074 static struct obd_ops filter_sanobd_ops = {
3075         .o_owner          = THIS_MODULE,
3076         .o_attach         = filter_attach,
3077         .o_detach         = filter_detach,
3078         .o_get_info       = filter_get_info,
3079         .o_setup          = filter_san_setup,
3080         .o_precleanup     = filter_precleanup,
3081         .o_cleanup        = filter_cleanup,
3082         .o_connect        = filter_connect,
3083         .o_connect_post   = filter_connect_post,
3084         .o_disconnect     = filter_disconnect,
3085         .o_statfs         = filter_statfs,
3086         .o_getattr        = filter_getattr,
3087         .o_unpackmd       = filter_unpackmd,
3088         .o_create         = filter_create,
3089         .o_setattr        = filter_setattr,
3090         .o_destroy        = filter_destroy,
3091         .o_brw            = filter_brw,
3092         .o_punch          = filter_truncate,
3093         .o_sync           = filter_sync,
3094         .o_preprw         = filter_preprw,
3095         .o_commitrw       = filter_commitrw,
3096         .o_do_cow         = filter_do_cow,
3097         .o_write_extents  = filter_write_extents,
3098         .o_san_preprw     = filter_san_preprw,
3099         .o_destroy_export = filter_destroy_export,
3100         .o_llog_init      = filter_llog_init,
3101         .o_llog_finish    = filter_llog_finish,
3102         .o_llog_connect   = filter_llog_connect,
3103         .o_iocontrol      = filter_iocontrol,
3104 };
3105
3106 static int __init obdfilter_init(void)
3107 {
3108         struct lprocfs_static_vars lvars;
3109         int size, rc;
3110
3111         printk(KERN_INFO "Lustre: Filtering OBD driver; info@clusterfs.com\n");
3112
3113         lprocfs_init_vars(filter, &lvars);
3114
3115         size = OBDFILTER_CREATED_SCRATCHPAD_ENTRIES *
3116                 sizeof(*obdfilter_created_scratchpad);
3117
3118         OBD_ALLOC(obdfilter_created_scratchpad, size);
3119         if (obdfilter_created_scratchpad == NULL) {
3120                 CERROR ("Can't allocate scratchpad\n");
3121                 return -ENOMEM;
3122         }
3123
3124         rc = class_register_type(&filter_obd_ops, NULL, lvars.module_vars,
3125                                  OBD_FILTER_DEVICENAME);
3126         if (rc) {
3127                 OBD_FREE(obdfilter_created_scratchpad, size);
3128                 return rc;
3129         }
3130
3131         rc = class_register_type(&filter_sanobd_ops, NULL, lvars.module_vars,
3132                                  OBD_FILTER_SAN_DEVICENAME);
3133         if (rc) {
3134                 class_unregister_type(OBD_FILTER_DEVICENAME);
3135                 OBD_FREE(obdfilter_created_scratchpad, size);
3136         }
3137         return rc;
3138 }
3139
3140 static void __exit obdfilter_exit(void)
3141 {
3142         class_unregister_type(OBD_FILTER_SAN_DEVICENAME);
3143         class_unregister_type(OBD_FILTER_DEVICENAME);
3144         OBD_FREE(obdfilter_created_scratchpad,
3145                  OBDFILTER_CREATED_SCRATCHPAD_ENTRIES *
3146                  sizeof(*obdfilter_created_scratchpad));
3147 }
3148
3149 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
3150 MODULE_DESCRIPTION("Lustre Filtering OBD driver");
3151 MODULE_LICENSE("GPL");
3152
3153 module_init(obdfilter_init);
3154 module_exit(obdfilter_exit);