Whamcloud - gitweb
- fixes and cleanups in error messages and in code.
[fs/lustre-release.git] / lustre / mds / handler.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  lustre/mds/handler.c
5  *  Lustre Metadata Server (mds) request handler
6  *
7  *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
8  *   Author: Peter Braam <braam@clusterfs.com>
9  *   Author: Andreas Dilger <adilger@clusterfs.com>
10  *   Author: Phil Schwan <phil@clusterfs.com>
11  *   Author: Mike Shaver <shaver@clusterfs.com>
12  *
13  *   This file is part of Lustre, http://www.lustre.org.
14  *
15  *   Lustre is free software; you can redistribute it and/or
16  *   modify it under the terms of version 2 of the GNU General Public
17  *   License as published by the Free Software Foundation.
18  *
19  *   Lustre is distributed in the hope that it will be useful,
20  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
21  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
22  *   GNU General Public License for more details.
23  *
24  *   You should have received a copy of the GNU General Public License
25  *   along with Lustre; if not, write to the Free Software
26  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
27  */
28
29 #ifndef EXPORT_SYMTAB
30 # define EXPORT_SYMTAB
31 #endif
32 #define DEBUG_SUBSYSTEM S_MDS
33
34 #include <linux/module.h>
35 #include <linux/lustre_mds.h>
36 #include <linux/lustre_dlm.h>
37 #include <linux/init.h>
38 #include <linux/obd_class.h>
39 #include <linux/random.h>
40 #include <linux/fs.h>
41 #include <linux/jbd.h>
42 #include <linux/ext3_fs.h>
43 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
44 # include <linux/smp_lock.h>
45 # include <linux/buffer_head.h>
46 # include <linux/workqueue.h>
47 # include <linux/mount.h>
48 #else
49 # include <linux/locks.h>
50 #endif
51 #include <linux/obd_lov.h>
52 #include <linux/obd_ost.h>
53 #include <linux/lustre_mds.h>
54 #include <linux/lustre_fsfilt.h>
55 #include <linux/lprocfs_status.h>
56 #include <linux/lustre_commit_confd.h>
57
58 #include "mds_internal.h"
59
60 static int mds_intent_policy(struct ldlm_namespace *ns,
61                              struct ldlm_lock **lockp, void *req_cookie,
62                              ldlm_mode_t mode, int flags, void *data);
63 static int mds_postsetup(struct obd_device *obd);
64 static int mds_cleanup(struct obd_device *obd, int flags);
65
66
67 /* Assumes caller has already pushed into the kernel filesystem context */
68 static int mds_sendpage(struct ptlrpc_request *req, struct file *file,
69                         loff_t offset, int count)
70 {
71         struct ptlrpc_bulk_desc *desc;
72         struct l_wait_info lwi;
73         struct page **pages;
74         int rc = 0, npages, i, tmpcount, tmpsize = 0;
75         ENTRY;
76
77         LASSERT((offset & (PAGE_SIZE - 1)) == 0); /* I'm dubious about this */
78
79         npages = (count + PAGE_SIZE - 1) >> PAGE_SHIFT;
80         OBD_ALLOC(pages, sizeof(*pages) * npages);
81         if (!pages)
82                 GOTO(out, rc = -ENOMEM);
83
84         desc = ptlrpc_prep_bulk_exp(req, npages, BULK_PUT_SOURCE,
85                                     MDS_BULK_PORTAL);
86         if (desc == NULL)
87                 GOTO(out_free, rc = -ENOMEM);
88
89         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
90                 tmpsize = tmpcount > PAGE_SIZE ? PAGE_SIZE : tmpcount;
91
92                 pages[i] = alloc_pages(GFP_KERNEL, 0);
93                 if (pages[i] == NULL)
94                         GOTO(cleanup_buf, rc = -ENOMEM);
95
96                 ptlrpc_prep_bulk_page(desc, pages[i], 0, tmpsize);
97         }
98
99         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
100                 tmpsize = tmpcount > PAGE_SIZE ? PAGE_SIZE : tmpcount;
101                 CDEBUG(D_EXT2, "reading %u@%llu from dir %lu (size %llu)\n",
102                        tmpsize, offset, file->f_dentry->d_inode->i_ino,
103                        file->f_dentry->d_inode->i_size);
104
105                 rc = fsfilt_readpage(req->rq_export->exp_obd, file,
106                                      kmap(pages[i]), tmpsize, &offset);
107                 kunmap(pages[i]);
108
109                 if (rc != tmpsize)
110                         GOTO(cleanup_buf, rc = -EIO);
111         }
112
113         LASSERT(desc->bd_nob == count);
114
115         rc = ptlrpc_start_bulk_transfer(desc);
116         if (rc)
117                 GOTO(cleanup_buf, rc);
118
119         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
120                 CERROR("obd_fail_loc=%x, fail operation rc=%d\n",
121                        OBD_FAIL_MDS_SENDPAGE, rc = -EIO);
122                 GOTO(abort_bulk, rc);
123         }
124
125         lwi = LWI_TIMEOUT(obd_timeout * HZ / 4, NULL, NULL);
126         rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc), &lwi);
127         LASSERT (rc == 0 || rc == -ETIMEDOUT);
128
129         if (rc == 0) {
130                 if (desc->bd_success &&
131                     desc->bd_nob_transferred == count)
132                         GOTO(cleanup_buf, rc);
133
134                 rc = -ETIMEDOUT; /* XXX should this be a different errno? */
135         }
136
137         DEBUG_REQ(D_ERROR, req, "bulk failed: %s %d(%d), evicting %s@%s\n",
138                   (rc == -ETIMEDOUT) ? "timeout" : "network error",
139                   desc->bd_nob_transferred, count,
140                   req->rq_export->exp_client_uuid.uuid,
141                   req->rq_export->exp_connection->c_remote_uuid.uuid);
142
143         ptlrpc_fail_export(req->rq_export);
144
145         EXIT;
146  abort_bulk:
147         ptlrpc_abort_bulk (desc);
148  cleanup_buf:
149         for (i = 0; i < npages; i++)
150                 if (pages[i])
151                         __free_pages(pages[i], 0);
152
153         ptlrpc_free_bulk(desc);
154  out_free:
155         OBD_FREE(pages, sizeof(*pages) * npages);
156  out:
157         return rc;
158 }
159
160 int mds_lock_mode_for_dir(struct obd_device *obd,
161                           struct dentry *dentry, int mode)
162 {
163         int ret_mode, split;
164
165         /* any dir access needs couple locks:
166          * 1) on part of dir we gonna lookup/modify in
167          * 2) on a whole dir to protect it from concurrent splitting
168          *    and to flush client's cache for readdir()
169          * so, for a given mode and dentry this routine decides what
170          * lock mode to use for lock #2:
171          * 1) if caller's gonna lookup in dir then we need to protect
172          *    dir from being splitted only - LCK_CR
173          * 2) if caller's gonna modify dir then we need to protect
174          *    dir from being splitted and to flush cache - LCK_CW
175          * 3) if caller's gonna modify dir and that dir seems ready
176          *    for splitting then we need to protect it from any
177          *    type of access (lookup/modify/split) - LCK_EX -bzzz */
178
179         split = mds_splitting_expected(obd, dentry);
180         
181         /*
182          * it is important to check here only for MDS_NO_SPLITTABLE. The reason
183          * is that MDS_NO_SPLITTABLE means dir is not splittable in principle
184          * and another thread will not split it on the quiet. But if we have
185          * MDS_NO_SPLIT_EXPECTED, this means, that dir may be splitted anytime,
186          * but not now (for current thread) and we should consider that it can
187          * happen soon and go that branch which can yield LCK_EX to protect from
188          * possible splitting.
189          */
190         if (split == MDS_NO_SPLITTABLE) {
191                 /*
192                  * this inode won't be splitted. so we need not to protect from
193                  * just flush client's cache on modification.
194                  */
195                 ret_mode = 0;
196                 if (mode == LCK_PW)
197                         ret_mode = LCK_CW;
198         } else {
199                 if (mode == LCK_EX)
200                         return LCK_EX;
201                 
202                 if (mode == LCK_PR) {
203                         ret_mode = LCK_CR;
204                 } else if (mode == LCK_PW) {
205                         /*
206                          * caller gonna modify directory.we use concurrent write
207                          * lock here to retract client's cache for readdir.
208                          */
209                         ret_mode = LCK_CW;
210                         if (split == MDS_EXPECT_SPLIT) {
211                                 /*
212                                  * splitting possible. serialize any access the
213                                  * idea is that first one seen dir is splittable
214                                  * is given exclusive lock and split
215                                  * directory. caller passes lock mode to
216                                  * mds_try_to_split_dir() and splitting would be
217                                  * done with exclusive lock only -bzzz.
218                                  */
219                                 CDEBUG(D_OTHER, "%s: gonna split %lu/%lu\n",
220                                        obd->obd_name,
221                                        (unsigned long)dentry->d_inode->i_ino,
222                                        (unsigned long)dentry->d_inode->i_generation);
223                                 ret_mode = LCK_EX;
224                         }
225                 }
226         }
227         return ret_mode;
228 }
229
230 /* only valid locked dentries or errors should be returned */
231 struct dentry *mds_id2locked_dentry(struct obd_device *obd, struct lustre_id *id,
232                                     struct vfsmount **mnt, int lock_mode,
233                                     struct lustre_handle *lockh, int *mode,
234                                     char *name, int namelen, __u64 lockpart)
235 {
236         struct dentry *de = mds_id2dentry(obd, id, mnt), *retval = de;
237         ldlm_policy_data_t policy = { .l_inodebits = { lockpart } };
238         struct ldlm_res_id res_id = { .name = {0} };
239         int flags = 0, rc;
240         ENTRY;
241
242         if (IS_ERR(de))
243                 RETURN(de);
244
245         lockh[1].cookie = 0;
246         res_id.name[0] = id_fid(id);
247         res_id.name[1] = id_group(id);
248         
249 #ifdef S_PDIROPS
250         if (name && IS_PDIROPS(de->d_inode)) {
251                 ldlm_policy_data_t cpolicy =
252                         { .l_inodebits = { MDS_INODELOCK_UPDATE } };
253                 LASSERT(mode != NULL);
254                 *mode = mds_lock_mode_for_dir(obd, de, lock_mode);
255                 if (*mode) {
256                         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace,
257                                               res_id, LDLM_IBITS,
258                                               &cpolicy, *mode, &flags,
259                                               mds_blocking_ast,
260                                               ldlm_completion_ast, NULL, NULL,
261                                               NULL, 0, NULL, lockh + 1);
262                         if (rc != ELDLM_OK) {
263                                 l_dput(de);
264                                 RETURN(ERR_PTR(-ENOLCK));
265                         }
266                 }
267                 flags = 0;
268
269                 res_id.name[2] = full_name_hash(name, namelen);
270
271                 CDEBUG(D_INFO, "take lock on "DLID4":"LPX64"\n",
272                        OLID4(id), res_id.name[2]);
273         }
274 #else
275 #warning "No PDIROPS support in the kernel"
276 #endif
277         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, res_id,
278                               LDLM_IBITS, &policy, lock_mode, &flags,
279                               mds_blocking_ast, ldlm_completion_ast,
280                               NULL, NULL, NULL, 0, NULL, lockh);
281         if (rc != ELDLM_OK) {
282                 l_dput(de);
283                 retval = ERR_PTR(-EIO); /* XXX translate ldlm code */
284 #ifdef S_PDIROPS
285                 if (lockh[1].cookie)
286                         ldlm_lock_decref(lockh + 1, LCK_CW);
287 #endif
288         }
289
290         RETURN(retval);
291 }
292
293 #ifndef DCACHE_DISCONNECTED
294 #define DCACHE_DISCONNECTED DCACHE_NFSD_DISCONNECTED
295 #endif
296
297
298 /* Look up an entry by inode number. This function ONLY returns valid dget'd
299  * dentries with an initialized inode or errors */
300 struct dentry *mds_id2dentry(struct obd_device *obd, struct lustre_id *id,
301                              struct vfsmount **mnt)
302 {
303         unsigned long ino = (unsigned long)id_ino(id);
304         __u32 generation = (__u32)id_gen(id);
305         struct mds_obd *mds = &obd->u.mds;
306         struct dentry *result;
307         struct inode *inode;
308         char idname[32];
309
310         if (ino == 0)
311                 RETURN(ERR_PTR(-ESTALE));
312
313         snprintf(idname, sizeof(idname), "0x%lx", ino);
314
315         CDEBUG(D_DENTRY, "--> mds_id2dentry: ino/gen %lu/%u, sb %p\n",
316                ino, generation, mds->mds_sb);
317
318         /* under ext3 this is neither supposed to return bad inodes nor NULL
319            inodes. */
320         result = ll_lookup_one_len(idname, mds->mds_id_de, 
321                                    strlen(idname));
322         if (IS_ERR(result))
323                 RETURN(result);
324
325         inode = result->d_inode;
326         if (!inode)
327                 RETURN(ERR_PTR(-ENOENT));
328
329         if (is_bad_inode(inode)) {
330                 CERROR("bad inode returned %lu/%u\n",
331                        inode->i_ino, inode->i_generation);
332                 dput(result);
333                 RETURN(ERR_PTR(-ENOENT));
334         }
335
336         /* here we disabled generation check, as root inode i_generation
337          * of cache mds and real mds are different. */
338         if (inode->i_ino != id_ino(&mds->mds_rootid) && generation &&
339             inode->i_generation != generation) {
340                 /* we didn't find the right inode.. */
341                 CERROR("bad inode %lu, link: %lu, ct: %d, generation %u/%u\n",
342                        inode->i_ino, (unsigned long)inode->i_nlink,
343                        atomic_read(&inode->i_count), inode->i_generation,
344                        generation);
345                 dput(result);
346                 RETURN(ERR_PTR(-ENOENT));
347         }
348
349         if (mnt) {
350                 *mnt = mds->mds_vfsmnt;
351                 mntget(*mnt);
352         }
353
354         RETURN(result);
355 }
356
357
358 /* Establish a connection to the MDS.
359  *
360  * This will set up an export structure for the client to hold state data about
361  * that client, like open files, the last operation number it did on the server,
362  * etc.
363  */
364 static int mds_connect(struct lustre_handle *conn, struct obd_device *obd,
365                        struct obd_uuid *cluuid, unsigned long flags)
366 {
367         struct mds_export_data *med;
368         struct mds_client_data *mcd;
369         struct obd_export *exp;
370         int rc;
371         ENTRY;
372
373         if (!conn || !obd || !cluuid)
374                 RETURN(-EINVAL);
375
376         /* XXX There is a small race between checking the list and adding a new
377          * connection for the same UUID, but the real threat (list corruption
378          * when multiple different clients connect) is solved.
379          *
380          * There is a second race between adding the export to the list, and
381          * filling in the client data below.  Hence skipping the case of NULL
382          * mcd above.  We should already be controlling multiple connects at the
383          * client, and we can't hold the spinlock over memory allocations
384          * without risk of deadlocking.
385          */
386         rc = class_connect(conn, obd, cluuid);
387         if (rc)
388                 RETURN(rc);
389         exp = class_conn2export(conn);
390         
391         LASSERT(exp != NULL);
392         med = &exp->exp_mds_data;
393
394         OBD_ALLOC(mcd, sizeof(*mcd));
395         if (!mcd) {
396                 CERROR("%s: out of memory for client data.\n",
397                         obd->obd_name);
398                 GOTO(out, rc = -ENOMEM);
399         }
400
401         memcpy(mcd->mcd_uuid, cluuid, sizeof(mcd->mcd_uuid));
402         med->med_mcd = mcd;
403
404         rc = mds_client_add(obd, &obd->u.mds, med, -1);
405         if (rc)
406                 GOTO(out, rc);
407        
408         EXIT;
409 out:
410         if (rc) {
411                 OBD_FREE(mcd, sizeof(*mcd));
412                 class_disconnect(exp, 0);
413         }
414         class_export_put(exp);
415         return rc;
416 }
417
418 static int mds_connect_post(struct obd_export *exp, unsigned long flags)
419 {
420         struct obd_device *obd = exp->exp_obd;
421         struct mds_obd *mds = &obd->u.mds;
422         int rc = 0;
423         ENTRY;
424
425         if (!(flags & OBD_OPT_MDS_CONNECTION)) {
426                 if (!(exp->exp_flags & OBD_OPT_REAL_CLIENT)) {
427                         atomic_inc(&mds->mds_real_clients);
428                         CDEBUG(D_OTHER,"%s: peer from %s is real client (%d)\n",
429                                obd->obd_name, exp->exp_client_uuid.uuid,
430                                atomic_read(&mds->mds_real_clients));
431                         exp->exp_flags |= OBD_OPT_REAL_CLIENT;
432                 }
433                 if (mds->mds_lmv_name)
434                         rc = mds_lmv_connect(obd, mds->mds_lmv_name);
435         }
436         RETURN(rc);
437 }
438
439 static int mds_init_export(struct obd_export *exp)
440 {
441         struct mds_export_data *med = &exp->exp_mds_data;
442
443         INIT_LIST_HEAD(&med->med_open_head);
444         spin_lock_init(&med->med_open_lock);
445         return 0;
446 }
447
448 static int mds_destroy_export(struct obd_export *export)
449 {
450         struct mds_export_data *med;
451         struct obd_device *obd = export->exp_obd;
452         struct lvfs_run_ctxt saved;
453         int rc = 0;
454         ENTRY;
455
456         med = &export->exp_mds_data;
457         target_destroy_export(export);
458
459         if (obd_uuid_equals(&export->exp_client_uuid, &obd->obd_uuid))
460                 GOTO(out, 0);
461
462         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
463
464         /* Close any open files (which may also cause orphan unlinking). */
465         spin_lock(&med->med_open_lock);
466         while (!list_empty(&med->med_open_head)) {
467                 struct list_head *tmp = med->med_open_head.next;
468                 struct mds_file_data *mfd =
469                         list_entry(tmp, struct mds_file_data, mfd_list);
470                 BDEVNAME_DECLARE_STORAGE(btmp);
471
472                 /* bug 1579: fix force-closing for 2.5 */
473                 struct dentry *dentry = mfd->mfd_dentry;
474
475                 list_del(&mfd->mfd_list);
476                 spin_unlock(&med->med_open_lock);
477
478                 /* If you change this message, be sure to update
479                  * replay_single:test_46 */
480                 CERROR("force closing client file handle for %*s (%s:%lu)\n",
481                        dentry->d_name.len, dentry->d_name.name,
482                        ll_bdevname(dentry->d_inode->i_sb, btmp),
483                        dentry->d_inode->i_ino);
484                 /* child inode->i_alloc_sem protects orphan_dec_test and
485                  * is_orphan race, mds_mfd_close drops it */
486                 DOWN_WRITE_I_ALLOC_SEM(dentry->d_inode);
487                 rc = mds_mfd_close(NULL, 0, obd, mfd,
488                                    !(export->exp_flags & OBD_OPT_FAILOVER));
489                 if (rc)
490                         CDEBUG(D_INODE, "Error closing file: %d\n", rc);
491                 spin_lock(&med->med_open_lock);
492         }
493         spin_unlock(&med->med_open_lock);
494         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
495
496         EXIT;
497 out:
498         mds_client_free(export, !(export->exp_flags & OBD_OPT_FAILOVER));
499         return rc;
500 }
501
502 static int mds_disconnect(struct obd_export *exp, unsigned long flags)
503 {
504         unsigned long irqflags;
505         struct obd_device *obd;
506         struct mds_obd *mds;
507         int rc;
508         ENTRY;
509
510         LASSERT(exp != NULL);
511         obd = class_exp2obd(exp);
512         if (obd == NULL) {
513                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
514                        exp->exp_handle.h_cookie);
515                 RETURN(-EINVAL);
516         }
517         mds = &obd->u.mds;
518
519         /*
520          * suppress any inter-mds requests durring disconnecting lmv if this is
521          * detected --force mode. This is needed to avoid endless recovery.
522          */
523         if (atomic_read(&mds->mds_real_clients) > 0 &&
524             !(exp->exp_flags & OBD_OPT_REAL_CLIENT))
525                 flags |= OBD_OPT_FORCE;
526                                                                                               
527         if (!(exp->exp_flags & OBD_OPT_REAL_CLIENT)
528             && !atomic_read(&mds->mds_real_clients)) {
529                 /* there was no client at all */
530                 mds_lmv_disconnect(obd, flags);
531         }
532
533         if ((exp->exp_flags & OBD_OPT_REAL_CLIENT)
534             && atomic_dec_and_test(&mds->mds_real_clients)) {
535                 /* time to drop LMV connections */
536                 CDEBUG(D_OTHER, "%s: last real client %s disconnected.  "
537                        "Disconnnect from LMV now\n",
538                        obd->obd_name, exp->exp_client_uuid.uuid);
539                 mds_lmv_disconnect(obd, flags);
540         }
541
542         spin_lock_irqsave(&exp->exp_lock, irqflags);
543         exp->exp_flags = flags;
544         spin_unlock_irqrestore(&exp->exp_lock, irqflags);
545
546         /* disconnect early so that clients can't keep using export */
547         rc = class_disconnect(exp, flags);
548         ldlm_cancel_locks_for_export(exp);
549
550         /* complete all outstanding replies */
551         spin_lock_irqsave(&exp->exp_lock, irqflags);
552         while (!list_empty(&exp->exp_outstanding_replies)) {
553                 struct ptlrpc_reply_state *rs =
554                         list_entry(exp->exp_outstanding_replies.next,
555                                    struct ptlrpc_reply_state, rs_exp_list);
556                 struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
557
558                 spin_lock(&svc->srv_lock);
559                 list_del_init(&rs->rs_exp_list);
560                 ptlrpc_schedule_difficult_reply(rs);
561                 spin_unlock(&svc->srv_lock);
562         }
563         spin_unlock_irqrestore(&exp->exp_lock, irqflags);
564         RETURN(rc);
565 }
566
567 static int mds_getstatus(struct ptlrpc_request *req)
568 {
569         struct mds_obd *mds = mds_req2mds(req);
570         struct mds_body *body;
571         int rc, size;
572         ENTRY;
573
574         size = sizeof(*body);
575         
576         rc = lustre_pack_reply(req, 1, &size, NULL);
577         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) {
578                 CERROR("mds: out of memory for message: size=%d\n", size);
579                 req->rq_status = -ENOMEM;       /* superfluous? */
580                 RETURN(-ENOMEM);
581         }
582
583         body = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*body));
584         
585         body->valid |= OBD_MD_FID;
586         memcpy(&body->id1, &mds->mds_rootid, sizeof(body->id1));
587
588         /*
589          * the last_committed and last_xid fields are filled in for all replies
590          * already - no need to do so here also.
591          */
592         RETURN(0);
593 }
594
595 int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
596                      void *data, int flag)
597 {
598         int do_ast;
599         ENTRY;
600
601         if (flag == LDLM_CB_CANCELING) {
602                 /* Don't need to do anything here. */
603                 RETURN(0);
604         }
605
606         /* XXX layering violation!  -phil */
607         l_lock(&lock->l_resource->lr_namespace->ns_lock);
608         
609         /*
610          * get this: if mds_blocking_ast is racing with mds_intent_policy, such
611          * that mds_blocking_ast is called just before l_i_p takes the ns_lock,
612          * then by the time we get the lock, we might not be the correct
613          * blocking function anymore.  So check, and return early, if so.
614          */
615         if (lock->l_blocking_ast != mds_blocking_ast) {
616                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
617                 RETURN(0);
618         }
619
620         lock->l_flags |= LDLM_FL_CBPENDING;
621         do_ast = (!lock->l_readers && !lock->l_writers);
622         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
623
624         if (do_ast) {
625                 struct lustre_handle lockh;
626                 int rc;
627
628                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
629                 ldlm_lock2handle(lock, &lockh);
630                 rc = ldlm_cli_cancel(&lockh);
631                 if (rc < 0)
632                         CERROR("ldlm_cli_cancel: %d\n", rc);
633         } else {
634                 LDLM_DEBUG(lock, "Lock still has references, will be "
635                            "cancelled later");
636         }
637         RETURN(0);
638 }
639
640 int mds_get_md(struct obd_device *obd, struct inode *inode, void *md,
641                int *size, int lock)
642 {
643         int lmm_size;
644         int rc = 0;
645         ENTRY;
646
647         if (lock)
648                 down(&inode->i_sem);
649         rc = fsfilt_get_md(obd, inode, md, *size);
650         if (lock)
651                 up(&inode->i_sem);
652
653         if (rc < 0) {
654                 CERROR("Error %d reading eadata for ino %lu\n",
655                        rc, inode->i_ino);
656         } else if (rc > 0) {
657                 lmm_size = rc;
658                 
659                 if (S_ISREG(inode->i_mode))
660                         rc = mds_convert_lov_ea(obd, inode, md, lmm_size);
661                 if (S_ISDIR(inode->i_mode))
662                         rc = mds_convert_mea_ea(obd, inode, md, lmm_size);
663
664                 if (rc == 0) {
665                         *size = lmm_size;
666                         rc = lmm_size;
667                 } else if (rc > 0) {
668                         *size = rc;
669                 }
670         }
671
672         RETURN (rc);
673 }
674
675
676 /* Call with lock=1 if you want mds_pack_md to take the i_sem.
677  * Call with lock=0 if the caller has already taken the i_sem. */
678 int mds_pack_md(struct obd_device *obd, struct lustre_msg *msg, int offset,
679                 struct mds_body *body, struct inode *inode, int lock)
680 {
681         struct mds_obd *mds = &obd->u.mds;
682         void *lmm;
683         int lmm_size;
684         int rc;
685         ENTRY;
686
687         lmm = lustre_msg_buf(msg, offset, 0);
688         if (lmm == NULL) {
689                 /* Some problem with getting eadata when I sized the reply
690                  * buffer... */
691                 CDEBUG(D_INFO, "no space reserved for inode %lu MD\n",
692                        inode->i_ino);
693                 RETURN(0);
694         }
695         lmm_size = msg->buflens[offset];
696
697         /* I don't really like this, but it is a sanity check on the client
698          * MD request.  However, if the client doesn't know how much space
699          * to reserve for the MD, it shouldn't be bad to have too much space.
700          */
701         if (lmm_size > mds->mds_max_mdsize) {
702                 CWARN("Reading MD for inode %lu of %d bytes > max %d\n",
703                        inode->i_ino, lmm_size, mds->mds_max_mdsize);
704                 // RETURN(-EINVAL);
705         }
706
707         rc = mds_get_md(obd, inode, lmm, &lmm_size, lock);
708         if (rc > 0) {
709                 if (S_ISDIR(inode->i_mode))
710                         body->valid |= OBD_MD_FLDIREA;
711                 else
712                         body->valid |= OBD_MD_FLEASIZE;
713                 body->eadatasize = lmm_size;
714                 rc = 0;
715         }
716
717         RETURN(rc);
718 }
719
720 void mds_squash_root(struct mds_obd *mds, struct mds_req_sec_desc *rsd,
721                      ptl_nid_t *peernid)
722 {
723         if (!mds->mds_squash_uid ||
724             (rsd->rsd_uid && rsd->rsd_fsuid))
725                 return;
726
727         if (*peernid == mds->mds_nosquash_nid)
728                 return;
729
730         CDEBUG(D_OTHER, "squash req from 0x%llx, (%d:%d/%x)=>(%d:%d/%x)\n",
731                 *peernid, rsd->rsd_fsuid, rsd->rsd_fsgid, rsd->rsd_cap,
732                 mds->mds_squash_uid, mds->mds_squash_gid,
733                 (rsd->rsd_cap & ~CAP_FS_MASK));
734
735         rsd->rsd_uid = mds->mds_squash_uid;
736         rsd->rsd_fsuid = mds->mds_squash_uid;
737         rsd->rsd_fsgid = mds->mds_squash_gid;
738
739         /* XXX should we remove all capabilities? */
740         rsd->rsd_cap &= ~CAP_FS_MASK;
741 }
742
743 static int mds_getattr_internal(struct obd_device *obd, struct dentry *dentry,
744                                 struct ptlrpc_request *req, struct mds_body *reqbody,
745                                 int reply_off)
746 {
747         struct inode *inode = dentry->d_inode;
748         struct mds_body *body;
749         int rc = 0;
750         ENTRY;
751
752         if (inode == NULL && !(dentry->d_flags & DCACHE_CROSS_REF))
753                 RETURN(-ENOENT);
754
755         body = lustre_msg_buf(req->rq_repmsg, reply_off, sizeof(*body));
756         LASSERT(body != NULL);                 /* caller prepped reply */
757
758         if (dentry->d_flags & DCACHE_CROSS_REF) {
759                 mds_pack_dentry2body(obd, body, dentry,
760                                      (reqbody->valid & OBD_MD_FID) ? 1 : 0);
761                 CDEBUG(D_OTHER, "cross reference: "DLID4"\n",
762                        OLID4(&body->id1));
763                 RETURN(0);
764         }
765         
766         mds_pack_inode2body(obd, body, inode, 
767                             (reqbody->valid & OBD_MD_FID) ? 1 : 0);
768
769         if ((S_ISREG(inode->i_mode) && (reqbody->valid & OBD_MD_FLEASIZE)) ||
770             (S_ISDIR(inode->i_mode) && (reqbody->valid & OBD_MD_FLDIREA))) {
771                 rc = mds_pack_md(obd, req->rq_repmsg, reply_off + 1, body,
772                                  inode, 1);
773
774                 /* if we have LOV EA data, the OST holds size, atime, mtime. */
775                 if (!(body->valid & OBD_MD_FLEASIZE) &&
776                     !(body->valid & OBD_MD_FLDIREA))
777                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
778                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
779         } else if (S_ISLNK(inode->i_mode) &&
780                    (reqbody->valid & OBD_MD_LINKNAME) != 0) {
781                 int len = req->rq_repmsg->buflens[reply_off + 1];
782                 char *symname = lustre_msg_buf(req->rq_repmsg, reply_off + 1, 0);
783
784                 LASSERT(symname != NULL);       /* caller prepped reply */
785
786                 if (!inode->i_op->readlink) {
787                         rc = -ENOSYS;
788                 } else {
789                         rc = inode->i_op->readlink(dentry, symname, len);
790                         if (rc < 0) {
791                                 CERROR("readlink failed: %d\n", rc);
792                         } else if (rc != len - 1) {
793                                 CERROR("Unexpected readlink rc %d: expecting %d\n",
794                                         rc, len - 1);
795                                 rc = -EINVAL;
796                         } else {
797                                 CDEBUG(D_INODE, "read symlink dest %s\n", symname);
798                                 body->valid |= OBD_MD_LINKNAME;
799                                 body->eadatasize = rc + 1;
800                                 symname[rc] = 0;
801                                 rc = 0;
802                         }
803                 }
804         }
805
806         RETURN(rc);
807 }
808
809 static int mds_getattr_pack_msg_cf(struct ptlrpc_request *req,
810                                    struct dentry *dentry,
811                                    int offset)
812 {
813         int rc = 0, size[1] = {sizeof(struct mds_body)};
814         ENTRY;
815
816         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
817                 CERROR("failed MDS_GETATTR_PACK test\n");
818                 req->rq_status = -ENOMEM;
819                 RETURN(-ENOMEM);
820         }
821
822         rc = lustre_pack_reply(req, 1, size, NULL);
823         if (rc) {
824                 CERROR("lustre_pack_reply failed: rc %d\n", rc);
825                 GOTO(out, req->rq_status = rc);
826         }
827
828         EXIT;
829 out:
830         return rc;
831 }
832
833 static int mds_getattr_pack_msg(struct ptlrpc_request *req, 
834                                 struct inode *inode,
835                                 int offset)
836 {
837         struct mds_obd *mds = mds_req2mds(req);
838         struct mds_body *body;
839         int rc = 0, size[2] = {sizeof(*body)}, bufcount = 1;
840         ENTRY;
841
842         body = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*body));
843         LASSERT(body != NULL);                 /* checked by caller */
844         LASSERT_REQSWABBED(req, offset);       /* swabbed by caller */
845
846         if ((S_ISREG(inode->i_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
847             (S_ISDIR(inode->i_mode) && (body->valid & OBD_MD_FLDIREA))) {
848                 int rc;
849                 down(&inode->i_sem);
850                 rc = fsfilt_get_md(req->rq_export->exp_obd, inode, NULL, 0);
851                 up(&inode->i_sem);
852                 CDEBUG(D_INODE, "got %d bytes MD data for inode %lu\n",
853                        rc, inode->i_ino);
854                 if (rc < 0) {
855                         if (rc != -ENODATA)
856                                 CERROR("error getting inode %lu MD: rc = %d\n",
857                                        inode->i_ino, rc);
858                         size[bufcount] = 0;
859                 } else if (rc > mds->mds_max_mdsize) {
860                         size[bufcount] = 0;
861                         CERROR("MD size %d larger than maximum possible %u\n",
862                                rc, mds->mds_max_mdsize);
863                 } else {
864                         size[bufcount] = rc;
865                 }
866                 bufcount++;
867         } else if (S_ISLNK(inode->i_mode) && (body->valid & OBD_MD_LINKNAME)) {
868                 if (inode->i_size + 1 != body->eadatasize)
869                         CERROR("symlink size: %Lu, reply space: %d\n",
870                                inode->i_size + 1, body->eadatasize);
871                 size[bufcount] = min_t(int, inode->i_size+1, body->eadatasize);
872                 bufcount++;
873                 CDEBUG(D_INODE, "symlink size: %Lu, reply space: %d\n",
874                        inode->i_size + 1, body->eadatasize);
875         }
876
877         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
878                 CERROR("failed MDS_GETATTR_PACK test\n");
879                 req->rq_status = -ENOMEM;
880                 GOTO(out, rc = -ENOMEM);
881         }
882
883         rc = lustre_pack_reply(req, bufcount, size, NULL);
884         if (rc) {
885                 CERROR("out of memory\n");
886                 GOTO(out, req->rq_status = rc);
887         }
888
889         EXIT;
890  out:
891         return rc;
892 }
893
894 int mds_check_mds_num(struct obd_device *obd, struct inode *inode,
895                       char *name, int namelen)
896 {
897         struct mea *mea = NULL;
898         int mea_size, rc = 0;
899         ENTRY;
900         
901         rc = mds_get_lmv_attr(obd, inode, &mea, &mea_size);
902         if (rc)
903                 RETURN(rc);
904         if (mea != NULL) {
905                 /*
906                  * dir is already splitted, check if requested filename should
907                  * live at this MDS or at another one.
908                  */
909                 int i = mea_name2idx(mea, name, namelen - 1);
910                 if (mea->mea_master != id_group(&mea->mea_ids[i])) {
911                         CDEBUG(D_OTHER,
912                                "inapropriate MDS(%d) for %s. should be "
913                                "%lu(%d)\n", mea->mea_master, name, 
914                                (unsigned long)id_group(&mea->mea_ids[i]), i);
915                         rc = -ERESTART;
916                 }
917         }
918
919         if (mea)
920                 OBD_FREE(mea, mea_size);
921         RETURN(rc);
922 }
923
924 static int mds_getattr_lock(struct ptlrpc_request *req, int offset,
925                             struct lustre_handle *child_lockh, int child_part)
926 {
927         struct obd_device *obd = req->rq_export->exp_obd;
928         struct mds_obd *mds = &obd->u.mds;
929         struct ldlm_reply *rep = NULL;
930         struct lvfs_run_ctxt saved;
931         struct mds_req_sec_desc *rsd;
932         struct mds_body *body;
933         struct dentry *dparent = NULL, *dchild = NULL;
934         struct lvfs_ucred uc;
935         struct lustre_handle parent_lockh[2] = {{0}, {0}};
936         unsigned int namesize;
937         int rc = 0, cleanup_phase = 0, resent_req = 0, update_mode, reply_offset;
938         char *name = NULL;
939         ENTRY;
940
941         LASSERT(!strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME));
942         MD_COUNTER_INCREMENT(obd, getattr_lock);
943
944         rsd = lustre_swab_mds_secdesc(req, MDS_REQ_SECDESC_OFF);
945         if (!rsd) {
946                 CERROR("Can't unpack security desc\n");
947                 RETURN(-EFAULT);
948         }
949         mds_squash_root(mds, rsd, &req->rq_peer.peer_id.nid); 
950
951         /* swab now, before anyone looks inside the request. */
952         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
953                                   lustre_swab_mds_body);
954         if (body == NULL) {
955                 CERROR("Can't swab mds_body\n");
956                 GOTO(cleanup, rc = -EFAULT);
957         }
958
959         LASSERT_REQSWAB(req, offset + 1);
960         name = lustre_msg_string(req->rq_reqmsg, offset + 1, 0);
961         if (name == NULL) {
962                 CERROR("Can't unpack name\n");
963                 GOTO(cleanup, rc = -EFAULT);
964         }
965         namesize = req->rq_reqmsg->buflens[offset + 1];
966
967         /* namesize less than 2 means we have empty name, probably came from
968            revalidate by cfid, so no point in having name to be set */
969         if (namesize <= 1)
970                 name = NULL;
971
972         LASSERT (offset == 1 || offset == 3);
973         if (offset == 3) {
974                 rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
975                 reply_offset = 1;
976         } else {
977                 reply_offset = 0;
978         }
979
980         rc = mds_init_ucred(&uc, rsd);
981         if (rc) {
982                 CERROR("can't init ucred\n");
983                 GOTO(cleanup, rc);
984         }
985
986         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
987         cleanup_phase = 1; /* kernel context */
988         intent_set_disposition(rep, DISP_LOOKUP_EXECD);
989
990         LASSERT(namesize > 0);
991         if (child_lockh->cookie != 0) {
992                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT);
993                 resent_req = 1;
994         }
995 #if 0        
996 #if HAVE_LOOKUP_RAW
997         if (body->valid == OBD_MD_FLID) {
998                 struct mds_body *mds_reply;
999                 int size = sizeof(*mds_reply);
1000                 struct inode *dir;
1001                 ino_t inum;
1002
1003                 dparent = mds_id2dentry(obd, &body->id1, NULL);
1004                 if (IS_ERR(dparent)) {
1005                         rc = PTR_ERR(dparent);
1006                         GOTO(cleanup, rc);
1007                 }
1008
1009                 /*
1010                  * the user requested ONLY the inode number, so do a raw lookup.
1011                  */
1012                 rc = lustre_pack_reply(req, 1, &size, NULL);
1013                 if (rc) {
1014                         CERROR("out of memory\n");
1015                         l_dput(dparent);
1016                         GOTO(cleanup, rc);
1017                 }
1018                 dir  = dparent->d_inode;
1019                 LASSERT(dir->i_op->lookup_raw != NULL);
1020                 rc = dir->i_op->lookup_raw(dir, name, namesize - 1, &inum);
1021                 l_dput(dparent);
1022                 mds_reply = lustre_msg_buf(req->rq_repmsg, 0,
1023                                            sizeof(*mds_reply));
1024
1025                 id_ino(&mds_reply->id1) = inum;
1026                 mds_reply->valid = OBD_MD_FLID;
1027                 GOTO(cleanup, rc);
1028         }
1029 #endif
1030 #endif
1031         if (resent_req == 0) {
1032                 LASSERT(id_fid(&body->id1) != 0);
1033                 if (name) {
1034                         rc = mds_get_parent_child_locked(obd, mds, &body->id1,
1035                                                          parent_lockh, &dparent,
1036                                                          LCK_PR, 
1037                                                          MDS_INODELOCK_UPDATE,
1038                                                          &update_mode, 
1039                                                          name, namesize,
1040                                                          child_lockh, &dchild, 
1041                                                          LCK_PR, child_part);
1042                         if (rc)
1043                                 GOTO(cleanup, rc);
1044                 
1045                         /*
1046                          * let's make sure this name should leave on this mds
1047                          * node.
1048                          */
1049                         rc = mds_check_mds_num(obd, dparent->d_inode, name, namesize);
1050                         if (rc)
1051                                 GOTO(cleanup, rc);
1052                 } else {
1053                         /* we have no dentry here, drop LOOKUP bit */
1054                         /* FIXME: we need MDS_INODELOCK_LOOKUP or not. */
1055                         child_part &= ~MDS_INODELOCK_LOOKUP;
1056                         CDEBUG(D_OTHER, "%s: retrieve attrs for "DLID4"\n",
1057                                obd->obd_name, OLID4(&body->id1));
1058
1059                         dchild = mds_id2locked_dentry(obd, &body->id1, NULL,
1060                                                       LCK_PR, parent_lockh,
1061                                                       &update_mode,
1062                                                       NULL, 0, 
1063                                                       MDS_INODELOCK_UPDATE);
1064                         if (IS_ERR(dchild)) {
1065                                 CERROR("can't find inode with id "DLID4", err = %d\n", 
1066                                        OLID4(&body->id1), (int)PTR_ERR(dchild));
1067                                 GOTO(cleanup, rc = PTR_ERR(dchild));
1068                         }
1069                         memcpy(child_lockh, parent_lockh, sizeof(parent_lockh[0]));
1070 #ifdef S_PDIROPS
1071                         if (parent_lockh[1].cookie)
1072                                 ldlm_lock_decref(parent_lockh + 1, update_mode);
1073 #endif
1074                 }
1075         } else {
1076                 struct ldlm_lock *granted_lock;
1077
1078                 DEBUG_REQ(D_DLMTRACE, req, "resent, not enqueuing new locks");
1079                 granted_lock = ldlm_handle2lock(child_lockh);
1080
1081                 LASSERTF(granted_lock != NULL, LPU64"/%lu lockh "LPX64"\n",
1082                          id_fid(&body->id1), (unsigned long)id_group(&body->id1),
1083                          child_lockh->cookie);
1084
1085                 dparent = mds_id2dentry(obd, &body->id1, NULL);
1086                 LASSERT(dparent);
1087
1088                 dchild = ll_lookup_one_len(name, dparent, namesize - 1);
1089                 LASSERT(dchild);
1090                 LDLM_LOCK_PUT(granted_lock);
1091         }
1092
1093         cleanup_phase = 2; /* dchild, dparent, locks */
1094
1095         if (!DENTRY_VALID(dchild)) {
1096                 intent_set_disposition(rep, DISP_LOOKUP_NEG);
1097                 /*
1098                  * in the intent case, the policy clears this error: the
1099                  * disposition is enough.
1100                  */
1101                 rc = -ENOENT;
1102                 GOTO(cleanup, rc);
1103         } else {
1104                 intent_set_disposition(rep, DISP_LOOKUP_POS);
1105         }
1106
1107         if (req->rq_repmsg == NULL) {
1108                 if (dchild->d_flags & DCACHE_CROSS_REF)
1109                         rc = mds_getattr_pack_msg_cf(req, dchild, offset);
1110                 else
1111                         rc = mds_getattr_pack_msg(req, dchild->d_inode, offset);
1112                 if (rc != 0) {
1113                         CERROR ("mds_getattr_pack_msg: %d\n", rc);
1114                         GOTO (cleanup, rc);
1115                 }
1116         }
1117
1118         rc = mds_getattr_internal(obd, dchild, req, body, reply_offset);
1119         GOTO(cleanup, rc); /* returns the lock to the client */
1120
1121  cleanup:
1122         switch (cleanup_phase) {
1123         case 2:
1124                 if (resent_req == 0) {
1125                         if (rc && DENTRY_VALID(dchild))
1126                                 ldlm_lock_decref(child_lockh, LCK_PR);
1127                         if (name)
1128                                 ldlm_lock_decref(parent_lockh, LCK_PR);
1129 #ifdef S_PDIROPS
1130                         if (parent_lockh[1].cookie != 0)
1131                                 ldlm_lock_decref(parent_lockh + 1, update_mode);
1132 #endif
1133                         if (dparent)
1134                                 l_dput(dparent);
1135                 }
1136                 l_dput(dchild);
1137         case 1:
1138                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1139                 mds_exit_ucred(&uc);
1140         }
1141         return rc;
1142 }
1143
1144 static int mds_getattr(struct ptlrpc_request *req, int offset)
1145 {
1146         struct obd_device *obd = req->rq_export->exp_obd;
1147         struct lvfs_run_ctxt saved;
1148         struct dentry *de;
1149         struct mds_req_sec_desc *rsd;
1150         struct mds_body *body;
1151         struct lvfs_ucred uc;
1152         int rc = 0;
1153         ENTRY;
1154
1155         rsd = lustre_swab_mds_secdesc(req, MDS_REQ_SECDESC_OFF);
1156         if (!rsd) {
1157                 CERROR("Can't unpack security desc\n");
1158                 RETURN(-EFAULT);
1159         }
1160
1161         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
1162                                   lustre_swab_mds_body);
1163         if (body == NULL) {
1164                 CERROR ("Can't unpack body\n");
1165                 RETURN (-EFAULT);
1166         }
1167
1168         MD_COUNTER_INCREMENT(obd, getattr);
1169
1170         rc = mds_init_ucred(&uc, rsd);
1171         if (rc) {
1172                 CERROR("can't init ucred\n");
1173                 RETURN(rc);
1174         }
1175
1176         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1177         de = mds_id2dentry(obd, &body->id1, NULL);
1178         if (IS_ERR(de)) {
1179                 rc = req->rq_status = PTR_ERR(de);
1180                 GOTO(out_pop, rc);
1181         }
1182
1183         rc = mds_getattr_pack_msg(req, de->d_inode, offset);
1184         if (rc != 0) {
1185                 CERROR("mds_getattr_pack_msg: %d\n", rc);
1186                 GOTO(out_pop, rc);
1187         }
1188
1189         req->rq_status = mds_getattr_internal(obd, de, req, body, 0);
1190
1191         l_dput(de);
1192
1193         EXIT;
1194 out_pop:
1195         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1196         mds_exit_ucred(&uc);
1197         return rc;
1198 }
1199
1200 static int mds_obd_statfs(struct obd_device *obd, struct obd_statfs *osfs,
1201                           unsigned long max_age)
1202 {
1203         int rc;
1204         ENTRY;
1205
1206         spin_lock(&obd->obd_osfs_lock);
1207         rc = fsfilt_statfs(obd, obd->u.mds.mds_sb, max_age);
1208         if (rc == 0)
1209                 memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
1210         spin_unlock(&obd->obd_osfs_lock);
1211
1212         RETURN(rc);
1213 }
1214
1215 static int mds_statfs(struct ptlrpc_request *req)
1216 {
1217         struct obd_device *obd = req->rq_export->exp_obd;
1218         int rc, size = sizeof(struct obd_statfs);
1219         ENTRY;
1220
1221         rc = lustre_pack_reply(req, 1, &size, NULL);
1222         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
1223                 CERROR("mds: statfs lustre_pack_reply failed: rc = %d\n", rc);
1224                 GOTO(out, rc);
1225         }
1226
1227         OBD_COUNTER_INCREMENT(obd, statfs);
1228
1229         /* We call this so that we can cache a bit - 1 jiffie worth */
1230         rc = mds_obd_statfs(obd, lustre_msg_buf(req->rq_repmsg, 0, size),
1231                             jiffies - HZ);
1232         if (rc) {
1233                 CERROR("mds_obd_statfs failed: rc %d\n", rc);
1234                 GOTO(out, rc);
1235         }
1236
1237         EXIT;
1238 out:
1239         req->rq_status = rc;
1240         return 0;
1241 }
1242
1243 static int mds_sync(struct ptlrpc_request *req, int offset)
1244 {
1245         struct obd_device *obd = req->rq_export->exp_obd;
1246         struct mds_obd *mds = &obd->u.mds;
1247         struct mds_body *body;
1248         int rc, size = sizeof(*body);
1249         ENTRY;
1250
1251         body = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*body));
1252         if (body == NULL)
1253                 GOTO(out, rc = -EPROTO);
1254
1255         rc = lustre_pack_reply(req, 1, &size, NULL);
1256         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK)) {
1257                 CERROR("fsync lustre_pack_reply failed: rc = %d\n", rc);
1258                 GOTO(out, rc);
1259         }
1260
1261         if (id_ino(&body->id1) == 0) {
1262                 /* an id of zero is taken to mean "sync whole filesystem" */
1263                 rc = fsfilt_sync(obd, mds->mds_sb);
1264                 if (rc)
1265                         GOTO(out, rc);
1266         } else {
1267                 /* just any file to grab fsync method - "file" arg unused */
1268                 struct file *file = mds->mds_rcvd_filp;
1269                 struct dentry *de;
1270
1271                 de = mds_id2dentry(obd, &body->id1, NULL);
1272                 if (IS_ERR(de))
1273                         GOTO(out, rc = PTR_ERR(de));
1274
1275                 rc = file->f_op->fsync(NULL, de, 1);
1276                 if (rc)
1277                         GOTO(out, rc);
1278
1279                 body = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*body));
1280                 mds_pack_inode2body(obd, body, de->d_inode, 0);
1281                 l_dput(de);
1282         }
1283
1284         EXIT;
1285 out:
1286         req->rq_status = rc;
1287         return 0;
1288 }
1289
1290 /* mds_readpage does not take a DLM lock on the inode, because the client must
1291  * already have a PR lock.
1292  *
1293  * If we were to take another one here, a deadlock will result, if another
1294  * thread is already waiting for a PW lock. */
1295 static int mds_readpage(struct ptlrpc_request *req, int offset)
1296 {
1297         struct obd_device *obd = req->rq_export->exp_obd;
1298         struct mds_obd *mds = &obd->u.mds;
1299         struct vfsmount *mnt;
1300         struct dentry *de;
1301         struct file *file;
1302         struct mds_req_sec_desc *rsd;
1303         struct mds_body *body, *repbody;
1304         struct lvfs_run_ctxt saved;
1305         int rc, size = sizeof(*repbody);
1306         struct lvfs_ucred uc;
1307         ENTRY;
1308
1309         rc = lustre_pack_reply(req, 1, &size, NULL);
1310         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK)) {
1311                 CERROR("mds: out of memory\n");
1312                 GOTO(out, rc = -ENOMEM);
1313         }
1314
1315         rsd = lustre_swab_mds_secdesc(req, MDS_REQ_SECDESC_OFF);
1316         if (!rsd) {
1317                 CERROR("Can't unpack security desc\n");
1318                 GOTO (out, rc = -EFAULT);
1319         }
1320         mds_squash_root(mds, rsd, &req->rq_peer.peer_id.nid); 
1321
1322         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
1323                                   lustre_swab_mds_body);
1324         if (body == NULL) {
1325                 CERROR("Can't unpack body\n");
1326                 GOTO (out, rc = -EFAULT);
1327         }
1328
1329         rc = mds_init_ucred(&uc, rsd);
1330         if (rc) {
1331                 CERROR("can't init ucred\n");
1332                 GOTO(out, rc);
1333         }
1334
1335         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1336         de = mds_id2dentry(obd, &body->id1, &mnt);
1337         if (IS_ERR(de))
1338                 GOTO(out_pop, rc = PTR_ERR(de));
1339
1340         CDEBUG(D_INODE, "ino %lu\n", de->d_inode->i_ino);
1341
1342         file = dentry_open(de, mnt, O_RDONLY | O_LARGEFILE);
1343         /* note: in case of an error, dentry_open puts dentry */
1344         if (IS_ERR(file))
1345                 GOTO(out_pop, rc = PTR_ERR(file));
1346
1347         /* body->size is actually the offset -eeb */
1348         if ((body->size & (de->d_inode->i_blksize - 1)) != 0) {
1349                 CERROR("offset "LPU64" not on a block boundary of %lu\n",
1350                        body->size, de->d_inode->i_blksize);
1351                 GOTO(out_file, rc = -EFAULT);
1352         }
1353
1354         /* body->nlink is actually the #bytes to read -eeb */
1355         if (body->nlink & (de->d_inode->i_blksize - 1)) {
1356                 CERROR("size %u is not multiple of blocksize %lu\n",
1357                        body->nlink, de->d_inode->i_blksize);
1358                 GOTO(out_file, rc = -EFAULT);
1359         }
1360
1361         repbody = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*repbody));
1362         repbody->size = file->f_dentry->d_inode->i_size;
1363         repbody->valid = OBD_MD_FLSIZE;
1364
1365         /* to make this asynchronous make sure that the handling function
1366            doesn't send a reply when this function completes. Instead a
1367            callback function would send the reply */
1368         /* body->size is actually the offset -eeb */
1369         rc = mds_sendpage(req, file, body->size, body->nlink);
1370
1371         EXIT;
1372 out_file:
1373         filp_close(file, 0);
1374 out_pop:
1375         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1376         mds_exit_ucred(&uc);
1377 out:
1378         req->rq_status = rc;
1379         return 0;
1380 }
1381
1382 /* update master MDS ID, which is stored in local inode EA. */
1383 int mds_update_mid(struct obd_device *obd, struct lustre_id *id,
1384                    void *data, int data_len)
1385 {
1386         struct mds_obd *mds = &obd->u.mds;
1387         struct dentry *dentry;
1388         void *handle;
1389         int rc = 0;
1390         ENTRY;
1391
1392         LASSERT(id);
1393         LASSERT(obd);
1394         
1395         dentry = mds_id2dentry(obd, id, NULL);
1396         if (IS_ERR(dentry))
1397                 GOTO(out, rc = PTR_ERR(dentry));
1398
1399         if (!dentry->d_inode) {
1400                 CERROR("Can't find object "DLID4".\n",
1401                        OLID4(id));
1402                 GOTO(out_dentry, rc = -EINVAL);
1403         }
1404
1405         handle = fsfilt_start(obd, dentry->d_inode,
1406                               FSFILT_OP_SETATTR, NULL);
1407         if (IS_ERR(handle))
1408                 GOTO(out_dentry, rc = PTR_ERR(handle));
1409
1410         rc = mds_update_inode_mid(obd, dentry->d_inode, handle,
1411                                   (struct lustre_id *)data);
1412         if (rc) {
1413                 CERROR("Can't update inode "DLID4" master id, "
1414                        "error = %d.\n", OLID4(id), rc);
1415                 GOTO(out_commit, rc);
1416         }
1417
1418         EXIT;
1419 out_commit:
1420         fsfilt_commit(obd, mds->mds_sb, dentry->d_inode,
1421                       handle, 0);
1422 out_dentry:
1423         l_dput(dentry);
1424 out:
1425         return rc;
1426 }
1427 EXPORT_SYMBOL(mds_update_mid);
1428
1429 /* read master MDS ID, which is stored in local inode EA. */
1430 int mds_read_mid(struct obd_device *obd, struct lustre_id *id,
1431                  void *data, int data_len)
1432 {
1433         struct dentry *dentry;
1434         int rc = 0;
1435         ENTRY;
1436
1437         LASSERT(id);
1438         LASSERT(obd);
1439         
1440         dentry = mds_id2dentry(obd, id, NULL);
1441         if (IS_ERR(dentry))
1442                 GOTO(out, rc = PTR_ERR(dentry));
1443
1444         if (!dentry->d_inode) {
1445                 CERROR("Can't find object "DLID4".\n",
1446                        OLID4(id));
1447                 GOTO(out_dentry, rc = -EINVAL);
1448         }
1449
1450         down(&dentry->d_inode->i_sem);
1451         rc = mds_read_inode_mid(obd, dentry->d_inode,
1452                                 (struct lustre_id *)data);
1453         up(&dentry->d_inode->i_sem);
1454         if (rc) {
1455                 CERROR("Can't read inode "DLID4" master id, "
1456                        "error = %d.\n", OLID4(id), rc);
1457                 GOTO(out_dentry, rc);
1458         }
1459
1460         EXIT;
1461 out_dentry:
1462         l_dput(dentry);
1463 out:
1464         return rc;
1465 }
1466 EXPORT_SYMBOL(mds_read_mid);
1467
1468 int mds_reint(struct ptlrpc_request *req, int offset,
1469               struct lustre_handle *lockh)
1470 {
1471         struct mds_obd *mds = &req->rq_export->exp_obd->u.mds;
1472         struct mds_update_record *rec;
1473         struct mds_req_sec_desc *rsd;
1474         int rc;
1475         ENTRY;
1476
1477         OBD_ALLOC(rec, sizeof(*rec));
1478         if (rec == NULL)
1479                 RETURN(-ENOMEM);
1480
1481         rsd = lustre_swab_mds_secdesc(req, MDS_REQ_SECDESC_OFF);
1482         if (!rsd) {
1483                 CERROR("Can't unpack security desc\n");
1484                 GOTO(out, rc = -EFAULT);
1485         }
1486         mds_squash_root(mds, rsd, &req->rq_peer.peer_id.nid); 
1487
1488         rc = mds_update_unpack(req, offset, rec);
1489         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) {
1490                 CERROR("invalid record\n");
1491                 GOTO(out, req->rq_status = -EINVAL);
1492         }
1493
1494         rc = mds_init_ucred(&rec->ur_uc, rsd);
1495         if (rc) {
1496                 CERROR("can't init ucred\n");
1497                 GOTO(out, rc);
1498         }
1499
1500         /* rc will be used to interrupt a for loop over multiple records */
1501         rc = mds_reint_rec(rec, offset, req, lockh);
1502         mds_exit_ucred(&rec->ur_uc);
1503         EXIT;
1504  out:
1505         OBD_FREE(rec, sizeof(*rec));
1506         return rc;
1507 }
1508
1509 static int mds_filter_recovery_request(struct ptlrpc_request *req,
1510                                        struct obd_device *obd, int *process)
1511 {
1512         switch (req->rq_reqmsg->opc) {
1513         case MDS_CONNECT: /* This will never get here, but for completeness. */
1514         case OST_CONNECT: /* This will never get here, but for completeness. */
1515         case MDS_DISCONNECT:
1516         case OST_DISCONNECT:
1517                *process = 1;
1518                RETURN(0);
1519
1520         case MDS_CLOSE:
1521         case MDS_SYNC: /* used in unmounting */
1522         case OBD_PING:
1523         case MDS_REINT:
1524         case LDLM_ENQUEUE:
1525         case OST_CREATE:
1526                 *process = target_queue_recovery_request(req, obd);
1527                 RETURN(0);
1528
1529         default:
1530                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1531                 *process = 0;
1532                 /* XXX what should we set rq_status to here? */
1533                 req->rq_status = -EAGAIN;
1534                 RETURN(ptlrpc_error(req));
1535         }
1536 }
1537
1538 static char *reint_names[] = {
1539         [REINT_SETATTR] "setattr",
1540         [REINT_CREATE]  "create",
1541         [REINT_LINK]    "link",
1542         [REINT_UNLINK]  "unlink",
1543         [REINT_RENAME]  "rename",
1544         [REINT_OPEN]    "open",
1545 };
1546
1547 #define FILTER_VALID_FLAGS (OBD_MD_FLTYPE | OBD_MD_FLMODE | OBD_MD_FLGENER  | \
1548                             OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ| \
1549                             OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME| \
1550                             OBD_MD_FLID) 
1551
1552 static void reconstruct_create(struct ptlrpc_request *req)
1553 {
1554         struct mds_export_data *med = &req->rq_export->exp_mds_data;
1555         struct mds_client_data *mcd = med->med_mcd;
1556         struct dentry *dentry;
1557         struct ost_body *body;
1558         struct lustre_id id;
1559         int rc;
1560         ENTRY;
1561
1562         /* copy rc, transno and disp; steal locks */
1563         mds_req_from_mcd(req, mcd);
1564         if (req->rq_status) {
1565                 EXIT;
1566                 return;
1567         }
1568
1569         id_gen(&id) = 0;
1570         id_group(&id) = 0;
1571
1572         id_ino(&id) = mcd->mcd_last_data;
1573         LASSERT(id_ino(&id) != 0);
1574
1575         dentry = mds_id2dentry(req2obd(req), &id, NULL);
1576         if (IS_ERR(dentry)) {
1577                 CERROR("can't find inode "LPU64"\n", id_ino(&id));
1578                 req->rq_status = PTR_ERR(dentry);
1579                 EXIT;
1580                 return;
1581         }
1582
1583         CWARN("reconstruct reply for x"LPU64" (remote ino) "LPU64" -> %lu/%u\n",
1584               req->rq_xid, id_ino(&id), dentry->d_inode->i_ino,
1585               dentry->d_inode->i_generation);
1586
1587         body = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*body));
1588         obdo_from_inode(&body->oa, dentry->d_inode, FILTER_VALID_FLAGS);
1589         body->oa.o_id = dentry->d_inode->i_ino;
1590         body->oa.o_generation = dentry->d_inode->i_generation;
1591         body->oa.o_valid |= OBD_MD_FLID | OBD_MD_FLGENER;
1592
1593         down(&dentry->d_inode->i_sem);
1594         rc = mds_read_inode_sid(req2obd(req), dentry->d_inode, &id);
1595         up(&dentry->d_inode->i_sem);
1596         if (rc) {
1597                 CERROR("Can't read inode self id, inode %lu, "
1598                        "rc %d\n", dentry->d_inode->i_ino, rc);
1599                 id_fid(&id) = 0;
1600         }
1601
1602         body->oa.o_fid = id_fid(&id);
1603         body->oa.o_mds = id_group(&id);
1604         l_dput(dentry);
1605
1606         EXIT;
1607 }
1608
1609 static int mdt_obj_create(struct ptlrpc_request *req)
1610 {
1611         struct obd_device *obd = req->rq_export->exp_obd;
1612         struct mds_obd *mds = &obd->u.mds;
1613         struct ost_body *body, *repbody;
1614         char idname[LL_ID_NAMELEN];
1615         int size = sizeof(*repbody);
1616         struct inode *parent_inode;
1617         struct lvfs_run_ctxt saved;
1618         int rc, cleanup_phase = 0;
1619         struct dentry *new = NULL;
1620         struct dentry_params dp;
1621         int mealen, flags = 0;
1622         struct lvfs_ucred uc;
1623         struct lustre_id id;
1624         struct mea *mea;
1625         void *handle = NULL;
1626         unsigned long cr_inum = 0;
1627         ENTRY;
1628        
1629         DEBUG_REQ(D_HA, req, "create remote object");
1630         parent_inode = mds->mds_unnamed_dir->d_inode;
1631
1632         body = lustre_swab_reqbuf(req, 0, sizeof(*body),
1633                                   lustre_swab_ost_body);
1634         if (body == NULL)
1635                 RETURN(-EFAULT);
1636
1637         rc = lustre_pack_reply(req, 1, &size, NULL);
1638         if (rc)
1639                 RETURN(rc);
1640
1641         MDS_CHECK_RESENT(req, reconstruct_create(req));
1642
1643         /*
1644          * this only serve to inter-mds request, don't need check group database
1645          * here. --ericm.
1646          */
1647         uc.luc_ghash = NULL;
1648         uc.luc_ginfo = NULL;
1649         uc.luc_uid = body->oa.o_uid;
1650         uc.luc_fsuid = body->oa.o_uid;
1651         uc.luc_fsgid = body->oa.o_gid;
1652
1653         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1654         repbody = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*repbody));
1655
1656         /* in REPLAY case inum should be given (client or other MDS fills it) */
1657         if (body->oa.o_id && ((body->oa.o_flags & OBD_FL_RECREATE_OBJS) ||
1658             (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY))) {
1659                 /*
1660                  * this is re-create request from MDS holding directory name.
1661                  * we have to lookup given ino/gen first. if it exists (good
1662                  * case) then there is nothing to do. if it does not then we
1663                  * have to recreate it.
1664                  */
1665                 id_ino(&id) = body->oa.o_id;
1666                 id_gen(&id) = body->oa.o_generation;
1667  
1668                 new = mds_id2dentry(obd, &id, NULL);
1669                 if (!IS_ERR(new) && new->d_inode) {
1670                         struct lustre_id sid;
1671                                 
1672                         CWARN("mkdir() repairing is on its way: %lu/%lu\n",
1673                               (unsigned long)id_ino(&id), (unsigned long)id_gen(&id));
1674                         
1675                         obdo_from_inode(&repbody->oa, new->d_inode,
1676                                         FILTER_VALID_FLAGS);
1677                         
1678                         repbody->oa.o_id = new->d_inode->i_ino;
1679                         repbody->oa.o_generation = new->d_inode->i_generation;
1680                         repbody->oa.o_valid |= OBD_MD_FLID | OBD_MD_FLGENER;
1681                         cleanup_phase = 1;
1682
1683                         down(&new->d_inode->i_sem);
1684                         rc = mds_read_inode_sid(obd, new->d_inode, &sid);
1685                         up(&new->d_inode->i_sem);
1686                         if (rc) {
1687                                 CERROR("Can't read inode self id "
1688                                        "inode %lu, rc %d.\n",
1689                                        new->d_inode->i_ino, rc);
1690                                 GOTO(cleanup, rc);
1691                         }
1692
1693                         repbody->oa.o_fid = id_fid(&sid);
1694                         repbody->oa.o_mds = id_group(&sid);
1695                         LASSERT(id_fid(&sid) != 0);
1696
1697                         /* 
1698                          * here we could use fid passed in body->oa.o_fid and
1699                          * thus avoid mds_read_inode_sid().
1700                          */
1701                         cr_inum = new->d_inode->i_ino;
1702                         GOTO(cleanup, rc = 0);
1703                 }
1704         }
1705         
1706         down(&parent_inode->i_sem);
1707         handle = fsfilt_start(obd, parent_inode, FSFILT_OP_MKDIR, NULL);
1708         if (IS_ERR(handle)) {
1709                 up(&parent_inode->i_sem);
1710                 CERROR("fsfilt_start() failed, rc = %d\n",
1711                        (int)PTR_ERR(handle));
1712                 GOTO(cleanup, rc = PTR_ERR(handle));
1713         }
1714         cleanup_phase = 1; /* transaction */
1715
1716 repeat:
1717         rc = sprintf(idname, "%u.%u", ll_insecure_random_int(), current->pid);
1718         new = lookup_one_len(idname, mds->mds_unnamed_dir, rc);
1719         if (IS_ERR(new)) {
1720                 CERROR("%s: can't lookup new inode (%s) for mkdir: %d\n",
1721                        obd->obd_name, idname, (int) PTR_ERR(new));
1722                 fsfilt_commit(obd, mds->mds_sb, new->d_inode, handle, 0);
1723                 up(&parent_inode->i_sem);
1724                 RETURN(PTR_ERR(new));
1725         } else if (new->d_inode) {
1726                 CERROR("%s: name exists. repeat\n", obd->obd_name);
1727                 goto repeat;
1728         }
1729
1730         new->d_fsdata = (void *)&dp;
1731         dp.p_inum = 0;
1732         dp.p_ptr = req;
1733
1734         if ((lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) ||
1735             (body->oa.o_flags & OBD_FL_RECREATE_OBJS)) {
1736                 LASSERT(body->oa.o_id != 0);
1737                 dp.p_inum = body->oa.o_id;
1738                 DEBUG_REQ(D_HA, req, "replay create obj %lu/%lu",
1739                           (unsigned long)body->oa.o_id,
1740                           (unsigned long)body->oa.o_generation);
1741         }
1742
1743         rc = vfs_mkdir(parent_inode, new, body->oa.o_mode);
1744         if (rc == 0) {
1745                 if ((body->oa.o_flags & OBD_FL_RECREATE_OBJS) ||
1746                     lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
1747                         new->d_inode->i_generation = body->oa.o_generation;
1748                         mark_inode_dirty(new->d_inode);
1749                         
1750                         /*
1751                          * avoiding asserts in cache flush case, as
1752                          * @body->oa.o_id should be zero.
1753                          */
1754                         if (body->oa.o_id) {
1755                                 LASSERTF(body->oa.o_id == new->d_inode->i_ino, 
1756                                          "BUG 3550: failed to recreate obj "
1757                                          LPU64" -> %lu\n", body->oa.o_id,
1758                                          new->d_inode->i_ino);
1759                                 
1760                                 LASSERTF(body->oa.o_generation == 
1761                                          new->d_inode->i_generation,
1762                                          "BUG 3550: failed to recreate obj/gen "
1763                                          LPU64"/%u -> %lu/%u\n", body->oa.o_id,
1764                                          body->oa.o_generation,
1765                                          new->d_inode->i_ino, 
1766                                          new->d_inode->i_generation);
1767                         }
1768                 }
1769                 
1770                 obdo_from_inode(&repbody->oa, new->d_inode, FILTER_VALID_FLAGS);
1771                 repbody->oa.o_id = new->d_inode->i_ino;
1772                 repbody->oa.o_generation = new->d_inode->i_generation;
1773                 repbody->oa.o_valid |= OBD_MD_FLID | OBD_MD_FLGENER | OBD_MD_FID;
1774
1775                 if ((body->oa.o_flags & OBD_FL_RECREATE_OBJS) ||
1776                     lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
1777                         id_group(&id) = mds->mds_num;
1778                 
1779                         LASSERT(body->oa.o_fid != 0);
1780                         id_fid(&id) = body->oa.o_fid;
1781
1782                         LASSERT(body->oa.o_id != 0);
1783                         id_ino(&id) = repbody->oa.o_id;
1784                         id_gen(&id) = repbody->oa.o_generation;
1785                 
1786                         down(&new->d_inode->i_sem);
1787                         rc = mds_update_inode_sid(obd, new->d_inode, handle, &id);
1788                         up(&new->d_inode->i_sem);
1789
1790                         /* 
1791                          * make sure, that fid is up-to-date.
1792                          */
1793                         mds_set_last_fid(obd, id_fid(&id));
1794                 } else {
1795                         /*
1796                          * allocate new sid, as object is created from scratch
1797                          * and this is not replay.
1798                          */
1799                         down(&new->d_inode->i_sem);
1800                         rc = mds_alloc_inode_sid(obd, new->d_inode, handle, &id);
1801                         up(&new->d_inode->i_sem);
1802                 }
1803                 if (rc) {
1804                         CERROR("Can't update lustre ID for inode %lu, "
1805                                "error = %d\n", new->d_inode->i_ino, rc);
1806                         GOTO(cleanup, rc);
1807                 }
1808
1809                 /* initializing o_fid after it is allocated. */
1810                 repbody->oa.o_fid = id_fid(&id);
1811                 repbody->oa.o_mds = id_group(&id);
1812
1813                 rc = fsfilt_del_dir_entry(obd, new);
1814                 up(&parent_inode->i_sem);
1815                 if (rc) {
1816                         CERROR("can't remove name for object: %d\n", rc);
1817                         GOTO(cleanup, rc);
1818                 }
1819                 
1820                 cleanup_phase = 2; /* created directory object */
1821
1822                 CDEBUG(D_OTHER, "created dirobj: %lu/%lu mode %o\n",
1823                        (unsigned long)new->d_inode->i_ino,
1824                        (unsigned long)new->d_inode->i_generation,
1825                        (unsigned)new->d_inode->i_mode);
1826                 cr_inum = new->d_inode->i_ino;
1827         } else {
1828                 up(&parent_inode->i_sem);
1829                 CERROR("%s: can't create dirobj: %d\n", obd->obd_name, rc);
1830                 GOTO(cleanup, rc);
1831         }
1832
1833         if (body->oa.o_valid & OBD_MD_FLID) {
1834                 /* this is new object for splitted dir. We have to prevent
1835                  * recursive splitting on it -bzzz */
1836                 mealen = obd_size_diskmd(mds->mds_lmv_exp, NULL);
1837
1838                 OBD_ALLOC(mea, mealen);
1839                 if (mea == NULL)
1840                         GOTO(cleanup, rc = -ENOMEM);
1841
1842                 mea->mea_magic = MEA_MAGIC_ALL_CHARS;
1843                 mea->mea_master = 0;
1844                 mea->mea_count = 0;
1845
1846                 down(&new->d_inode->i_sem);
1847                 rc = fsfilt_set_md(obd, new->d_inode, handle, mea, mealen);
1848                 up(&new->d_inode->i_sem);
1849                 if (rc)
1850                         CERROR("fsfilt_set_md() failed, rc = %d\n", rc);
1851
1852                 OBD_FREE(mea, mealen);
1853                 CDEBUG(D_OTHER, "%s: mark non-splittable %lu/%u - %d\n",
1854                        obd->obd_name, new->d_inode->i_ino,
1855                        new->d_inode->i_generation, flags);
1856         } else if (body->oa.o_easize) {
1857                 /* we pass LCK_EX to split routine to signal that we have
1858                  * exclusive access to the directory. simple because nobody
1859                  * knows it already exists -bzzz */
1860                 rc = mds_try_to_split_dir(obd, new, NULL,
1861                                           body->oa.o_easize, LCK_EX);
1862                 if (rc < 0) {
1863                         CERROR("Can't split directory %lu, error = %d.\n",
1864                                new->d_inode->i_ino, rc);
1865                 } else {
1866                         rc = 0;
1867                 }
1868         }
1869
1870         EXIT;
1871 cleanup:
1872         switch (cleanup_phase) {
1873         case 2: /* object has been created, but we'll may want to replay it later */
1874                 if (rc == 0)
1875                         ptlrpc_require_repack(req);
1876         case 1: /* transaction */
1877                 rc = mds_finish_transno(mds, parent_inode, handle,
1878                                         req, rc, cr_inum);
1879         }
1880
1881         l_dput(new);
1882         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1883         mds_put_group_entry(mds, uc.luc_ghash);
1884         return rc;
1885 }
1886
1887 static int mdt_get_info(struct ptlrpc_request *req)
1888 {
1889         struct obd_export *exp = req->rq_export;
1890         int keylen, rc = 0;
1891         char *key;
1892         ENTRY;
1893
1894         key = lustre_msg_buf(req->rq_reqmsg, 0, 1);
1895         if (key == NULL) {
1896                 DEBUG_REQ(D_HA, req, "no get_info key");
1897                 RETURN(-EFAULT);
1898         }
1899         keylen = req->rq_reqmsg->buflens[0];
1900
1901         if ((keylen < strlen("mdsize") || strcmp(key, "mdsize") != 0) &&
1902             (keylen < strlen("mdsnum") || strcmp(key, "mdsnum") != 0) &&
1903             (keylen < strlen("rootid") || strcmp(key, "rootid") != 0))
1904                 RETURN(-EPROTO);
1905
1906         if (keylen >= strlen("rootid") && !strcmp(key, "rootid")) {
1907                 struct lustre_id *reply;
1908                 int size = sizeof(*reply);
1909                 
1910                 rc = lustre_pack_reply(req, 1, &size, NULL);
1911                 if (rc)
1912                         RETURN(rc);
1913
1914                 reply = lustre_msg_buf(req->rq_repmsg, 0, size);
1915                 rc = obd_get_info(exp, keylen, key, &size, reply);
1916         } else {
1917                 obd_id *reply;
1918                 int size = sizeof(*reply);
1919                 
1920                 rc = lustre_pack_reply(req, 1, &size, NULL);
1921                 if (rc)
1922                         RETURN(rc);
1923
1924                 reply = lustre_msg_buf(req->rq_repmsg, 0, size);
1925                 rc = obd_get_info(exp, keylen, key, &size, reply);
1926         }
1927
1928         req->rq_repmsg->status = 0;
1929         RETURN(rc);
1930 }
1931
1932 static int mds_set_info(struct obd_export *exp, __u32 keylen,
1933                         void *key, __u32 vallen, void *val)
1934 {
1935         struct obd_device *obd;
1936         struct mds_obd *mds;
1937         int rc = 0;
1938         ENTRY;
1939
1940         obd = class_exp2obd(exp);
1941         if (obd == NULL) {
1942                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
1943                        exp->exp_handle.h_cookie);
1944                 RETURN(-EINVAL);
1945         }
1946
1947         mds = &obd->u.mds;
1948         if (keylen >= strlen("mds_type") &&
1949              memcmp(key, "mds_type", keylen) == 0) {
1950                 int valsize;
1951                 __u32 group;
1952                 
1953                 CDEBUG(D_IOCTL, "set mds type to %x\n", *(int*)val);
1954                 
1955                 mds->mds_obd_type = *(int*)val;
1956                 group = FILTER_GROUP_FIRST_MDS + mds->mds_obd_type;
1957                 valsize = sizeof(group);
1958                 
1959                 /* mds number has been changed, so the corresponding obdfilter
1960                  * exp need to be changed too. */
1961                 rc = obd_set_info(mds->mds_lov_exp, strlen("mds_conn"),
1962                                   "mds_conn", valsize, &group);
1963                 RETURN(rc);
1964         }
1965         CDEBUG(D_IOCTL, "invalid key\n");
1966         RETURN(-EINVAL);
1967 }
1968
1969 static int mdt_set_info(struct ptlrpc_request *req)
1970 {
1971         char *key, *val;
1972         struct obd_export *exp = req->rq_export;
1973         int keylen, rc = 0, vallen;
1974         ENTRY;
1975
1976         key = lustre_msg_buf(req->rq_reqmsg, 0, 1);
1977         if (key == NULL) {
1978                 DEBUG_REQ(D_HA, req, "no set_info key");
1979                 RETURN(-EFAULT);
1980         }
1981         keylen = req->rq_reqmsg->buflens[0];
1982
1983         if (keylen == strlen("mds_type") &&
1984             memcmp(key, "mds_type", keylen) == 0) {
1985                 rc = lustre_pack_reply(req, 0, NULL, NULL);
1986                 if (rc)
1987                         RETURN(rc);
1988                 
1989                 val = lustre_msg_buf(req->rq_reqmsg, 1, 0);
1990                 vallen = req->rq_reqmsg->buflens[1];
1991
1992                 rc = obd_set_info(exp, keylen, key, vallen, val);
1993                 req->rq_repmsg->status = 0;
1994                 RETURN(rc);
1995         }
1996         CDEBUG(D_IOCTL, "invalid key\n");
1997         RETURN(-EINVAL);
1998 }
1999
2000 static int mds_msg_check_version(struct lustre_msg *msg)
2001 {
2002         int rc;
2003
2004         switch (msg->opc) {
2005         case MDS_CONNECT:
2006         case MDS_DISCONNECT:
2007         case OBD_PING:
2008                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
2009                 if (rc)
2010                         CERROR("bad opc %u version %08x, expecting %08x\n",
2011                                msg->opc, msg->version, LUSTRE_OBD_VERSION);
2012                 break;
2013         case MDS_STATFS:
2014         case MDS_GETSTATUS:
2015         case MDS_GETATTR:
2016         case MDS_GETATTR_LOCK:
2017         case MDS_READPAGE:
2018         case MDS_REINT:
2019         case MDS_CLOSE:
2020         case MDS_DONE_WRITING:
2021         case MDS_PIN:
2022         case MDS_SYNC:
2023                 rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
2024                 if (rc)
2025                         CERROR("bad opc %u version %08x, expecting %08x\n",
2026                                msg->opc, msg->version, LUSTRE_MDS_VERSION);
2027                 break;
2028         case LDLM_ENQUEUE:
2029         case LDLM_CONVERT:
2030         case LDLM_BL_CALLBACK:
2031         case LDLM_CP_CALLBACK:
2032                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
2033                 if (rc)
2034                         CERROR("bad opc %u version %08x, expecting %08x\n",
2035                                msg->opc, msg->version, LUSTRE_DLM_VERSION);
2036                 break;
2037         case OBD_LOG_CANCEL:
2038         case LLOG_ORIGIN_HANDLE_OPEN:
2039         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
2040         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
2041         case LLOG_ORIGIN_HANDLE_READ_HEADER:
2042         case LLOG_ORIGIN_HANDLE_CLOSE:
2043         case LLOG_CATINFO:
2044                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
2045                 if (rc)
2046                         CERROR("bad opc %u version %08x, expecting %08x\n",
2047                                msg->opc, msg->version, LUSTRE_LOG_VERSION);
2048                 break;
2049         case OST_CREATE:
2050         case OST_WRITE:
2051         case OST_GET_INFO:
2052         case OST_SET_INFO:
2053                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
2054                 if (rc)
2055                         CERROR("bad opc %u version %08x, expecting %08x\n",
2056                                msg->opc, msg->version, LUSTRE_OBD_VERSION);
2057                 break;
2058         default:
2059                 CERROR("MDS unknown opcode %d\n", msg->opc);
2060                 rc = -ENOTSUPP;
2061                 break;
2062         }
2063
2064         return rc;
2065 }
2066
2067 static char str[PTL_NALFMT_SIZE];
2068
2069 int mds_handle(struct ptlrpc_request *req)
2070 {
2071         int should_process, fail = OBD_FAIL_MDS_ALL_REPLY_NET;
2072         int rc = 0;
2073         struct mds_obd *mds = NULL; /* quell gcc overwarning */
2074         struct obd_device *obd = NULL;
2075         ENTRY;
2076
2077         OBD_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
2078
2079         rc = mds_msg_check_version(req->rq_reqmsg);
2080         if (rc) {
2081                 CERROR("MDS drop mal-formed request\n");
2082                 RETURN(rc);
2083         }
2084
2085         LASSERT(current->journal_info == NULL);
2086         /* XXX identical to OST */
2087         if (req->rq_reqmsg->opc != MDS_CONNECT) {
2088                 struct mds_export_data *med;
2089                 int recovering;
2090
2091                 if (req->rq_export == NULL) {
2092                         CERROR("operation %d on unconnected MDS from NID %s\n",
2093                                req->rq_reqmsg->opc,
2094                                ptlrpc_peernid2str(&req->rq_peer, str));
2095                         req->rq_status = -ENOTCONN;
2096                         GOTO(out, rc = -ENOTCONN);
2097                 }
2098
2099                 med = &req->rq_export->exp_mds_data;
2100                 obd = req->rq_export->exp_obd;
2101                 mds = &obd->u.mds;
2102
2103                 /* sanity check: if the xid matches, the request must
2104                  * be marked as a resent or replayed */
2105                 if (req->rq_xid == med->med_mcd->mcd_last_xid) {
2106                         LASSERTF(lustre_msg_get_flags(req->rq_reqmsg) &
2107                                  (MSG_RESENT | MSG_REPLAY),
2108                                  "rq_xid "LPU64" matches last_xid, "
2109                                  "expected RESENT flag\n",
2110                                  req->rq_xid);
2111                 }
2112                 /* else: note the opposite is not always true; a
2113                  * RESENT req after a failover will usually not match
2114                  * the last_xid, since it was likely never
2115                  * committed. A REPLAYed request will almost never
2116                  * match the last xid, however it could for a
2117                  * committed, but still retained, open. */
2118
2119                 spin_lock_bh(&obd->obd_processing_task_lock);
2120                 recovering = obd->obd_recovering;
2121                 spin_unlock_bh(&obd->obd_processing_task_lock);
2122                 if (recovering) {
2123                         rc = mds_filter_recovery_request(req, obd,
2124                                                          &should_process);
2125                         if (rc || should_process == 0) {
2126                                 RETURN(rc);
2127                         } else if (should_process < 0) {
2128                                 req->rq_status = should_process;
2129                                 rc = ptlrpc_error(req);
2130                                 RETURN(rc);
2131                         }
2132                 }
2133         }
2134
2135         switch (req->rq_reqmsg->opc) {
2136         case MDS_CONNECT:
2137                 DEBUG_REQ(D_INODE, req, "connect");
2138                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CONNECT_NET, 0);
2139                 rc = target_handle_connect(req);
2140                 if (!rc)
2141                         /* Now that we have an export, set mds. */
2142                         mds = mds_req2mds(req);
2143                 break;
2144
2145         case MDS_DISCONNECT:
2146                 DEBUG_REQ(D_INODE, req, "disconnect");
2147                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DISCONNECT_NET, 0);
2148                 rc = target_handle_disconnect(req);
2149                 req->rq_status = rc;            /* superfluous? */
2150                 break;
2151
2152         case MDS_GETSTATUS:
2153                 DEBUG_REQ(D_INODE, req, "getstatus");
2154                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_NET, 0);
2155                 rc = mds_getstatus(req);
2156                 break;
2157
2158         case MDS_GETATTR:
2159                 DEBUG_REQ(D_INODE, req, "getattr");
2160                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NET, 0);
2161                 rc = mds_getattr(req, MDS_REQ_REC_OFF);
2162                 break;
2163
2164         case MDS_GETATTR_LOCK: {
2165                 struct lustre_handle lockh;
2166                 DEBUG_REQ(D_INODE, req, "getattr_lock");
2167                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_LOCK_NET, 0);
2168
2169                 /* If this request gets a reconstructed reply, we won't be
2170                  * acquiring any new locks in mds_getattr_lock, so we don't
2171                  * want to cancel.
2172                  */
2173                 lockh.cookie = 0;
2174                 rc = mds_getattr_lock(req, MDS_REQ_REC_OFF, &lockh,
2175                                       MDS_INODELOCK_UPDATE);
2176                 /* this non-intent call (from an ioctl) is special */
2177                 req->rq_status = rc;
2178                 if (rc == 0 && lockh.cookie)
2179                         ldlm_lock_decref(&lockh, LCK_PR);
2180                 break;
2181         }
2182         case MDS_STATFS:
2183                 DEBUG_REQ(D_INODE, req, "statfs");
2184                 OBD_FAIL_RETURN(OBD_FAIL_MDS_STATFS_NET, 0);
2185                 rc = mds_statfs(req);
2186                 break;
2187
2188         case MDS_READPAGE:
2189                 DEBUG_REQ(D_INODE, req, "readpage");
2190                 OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_NET, 0);
2191                 rc = mds_readpage(req, MDS_REQ_REC_OFF);
2192
2193                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_SENDPAGE)) {
2194                         if (req->rq_reply_state) {
2195                                 lustre_free_reply_state (req->rq_reply_state);
2196                                 req->rq_reply_state = NULL;
2197                         }
2198                         RETURN(0);
2199                 }
2200
2201                 break;
2202         case MDS_REINT: {
2203                 __u32 *opcp = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF,
2204                                              sizeof (*opcp));
2205                 __u32  opc;
2206                 int size[3] = {sizeof(struct mds_body), mds->mds_max_mdsize,
2207                                mds->mds_max_cookiesize};
2208                 int bufcount;
2209
2210                 /* NB only peek inside req now; mds_reint() will swab it */
2211                 if (opcp == NULL) {
2212                         CERROR ("Can't inspect opcode\n");
2213                         rc = -EINVAL;
2214                         break;
2215                 }
2216                 opc = *opcp;
2217                 if (lustre_msg_swabbed (req->rq_reqmsg))
2218                         __swab32s(&opc);
2219
2220                 DEBUG_REQ(D_INODE, req, "reint %d (%s)", opc,
2221                           (opc < sizeof(reint_names) / sizeof(reint_names[0]) ||
2222                            reint_names[opc] == NULL) ? reint_names[opc] :
2223                                                        "unknown opcode");
2224
2225                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
2226
2227                 if (opc == REINT_UNLINK || opc == REINT_RENAME)
2228                         bufcount = 3;
2229                 else if (opc == REINT_OPEN)
2230                         bufcount = 2;
2231                 else
2232                         bufcount = 1;
2233
2234                 rc = lustre_pack_reply(req, bufcount, size, NULL);
2235                 if (rc)
2236                         break;
2237
2238                 rc = mds_reint(req, MDS_REQ_REC_OFF, NULL);
2239                 fail = OBD_FAIL_MDS_REINT_NET_REP;
2240                 break;
2241         }
2242
2243         case MDS_CLOSE:
2244                 DEBUG_REQ(D_INODE, req, "close");
2245                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CLOSE_NET, 0);
2246                 rc = mds_close(req, MDS_REQ_REC_OFF);
2247                 break;
2248
2249         case MDS_DONE_WRITING:
2250                 DEBUG_REQ(D_INODE, req, "done_writing");
2251                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DONE_WRITING_NET, 0);
2252                 rc = mds_done_writing(req, MDS_REQ_REC_OFF);
2253                 break;
2254
2255         case MDS_PIN:
2256                 DEBUG_REQ(D_INODE, req, "pin");
2257                 OBD_FAIL_RETURN(OBD_FAIL_MDS_PIN_NET, 0);
2258                 rc = mds_pin(req, MDS_REQ_REC_OFF);
2259                 break;
2260
2261         case MDS_SYNC:
2262                 DEBUG_REQ(D_INODE, req, "sync");
2263                 OBD_FAIL_RETURN(OBD_FAIL_MDS_SYNC_NET, 0);
2264                 rc = mds_sync(req, MDS_REQ_REC_OFF);
2265                 break;
2266
2267         case OBD_PING:
2268                 DEBUG_REQ(D_INODE, req, "ping");
2269                 rc = target_handle_ping(req);
2270                 break;
2271
2272         case OBD_LOG_CANCEL:
2273                 CDEBUG(D_INODE, "log cancel\n");
2274                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
2275                 rc = -ENOTSUPP; /* la la la */
2276                 break;
2277
2278         case LDLM_ENQUEUE:
2279                 DEBUG_REQ(D_INODE, req, "enqueue");
2280                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
2281                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
2282                                          ldlm_server_blocking_ast, NULL);
2283                 fail = OBD_FAIL_LDLM_REPLY;
2284                 break;
2285         case LDLM_CONVERT:
2286                 DEBUG_REQ(D_INODE, req, "convert");
2287                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
2288                 rc = ldlm_handle_convert(req);
2289                 break;
2290         case LDLM_BL_CALLBACK:
2291         case LDLM_CP_CALLBACK:
2292                 DEBUG_REQ(D_INODE, req, "callback");
2293                 CERROR("callbacks should not happen on MDS\n");
2294                 LBUG();
2295                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
2296                 break;
2297         case LLOG_ORIGIN_HANDLE_OPEN:
2298                 DEBUG_REQ(D_INODE, req, "llog_init");
2299                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2300                 rc = llog_origin_handle_open(req);
2301                 break;
2302         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
2303                 DEBUG_REQ(D_INODE, req, "llog next block");
2304                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2305                 rc = llog_origin_handle_next_block(req);
2306                 break;
2307         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
2308                 DEBUG_REQ(D_INODE, req, "llog prev block");
2309                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2310                 rc = llog_origin_handle_prev_block(req);
2311                 break;
2312         case LLOG_ORIGIN_HANDLE_READ_HEADER:
2313                 DEBUG_REQ(D_INODE, req, "llog read header");
2314                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2315                 rc = llog_origin_handle_read_header(req);
2316                 break;
2317         case LLOG_ORIGIN_HANDLE_CLOSE:
2318                 DEBUG_REQ(D_INODE, req, "llog close");
2319                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2320                 rc = llog_origin_handle_close(req);
2321                 break;
2322         case OST_CREATE:
2323                 DEBUG_REQ(D_INODE, req, "ost_create");
2324                 rc = mdt_obj_create(req);
2325                 break;
2326         case OST_GET_INFO:
2327                 DEBUG_REQ(D_INODE, req, "get_info");
2328                 rc = mdt_get_info(req);
2329                 break;
2330         case OST_SET_INFO:
2331                 DEBUG_REQ(D_INODE, req, "set_info");
2332                 rc = mdt_set_info(req);
2333                 break;
2334         case OST_WRITE:
2335                 CDEBUG(D_INODE, "write\n");
2336                 OBD_FAIL_RETURN(OBD_FAIL_OST_BRW_NET, 0);
2337                 rc = ost_brw_write(req, NULL);
2338                 LASSERT(current->journal_info == NULL);
2339                 /* mdt_brw sends its own replies */
2340                 RETURN(rc);
2341                 break;
2342         case LLOG_CATINFO:
2343                 DEBUG_REQ(D_INODE, req, "llog catinfo");
2344                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
2345                 rc = llog_catinfo(req);
2346                 break;
2347         default:
2348                 req->rq_status = -ENOTSUPP;
2349                 rc = ptlrpc_error(req);
2350                 RETURN(rc);
2351         }
2352
2353         LASSERT(current->journal_info == NULL);
2354
2355         EXIT;
2356
2357         /* If we're DISCONNECTing, the mds_export_data is already freed */
2358         if (!rc && req->rq_reqmsg->opc != MDS_DISCONNECT) {
2359                 struct mds_export_data *med = &req->rq_export->exp_mds_data;
2360                 struct obd_device *obd = list_entry(mds, struct obd_device,
2361                                                     u.mds);
2362                 req->rq_repmsg->last_xid =
2363                         le64_to_cpu(med->med_mcd->mcd_last_xid);
2364
2365                 if (!obd->obd_no_transno) {
2366                         req->rq_repmsg->last_committed =
2367                                 obd->obd_last_committed;
2368                 } else {
2369                         DEBUG_REQ(D_IOCTL, req,
2370                                   "not sending last_committed update");
2371                 }
2372                 CDEBUG(D_INFO, "last_transno "LPU64", last_committed "LPU64
2373                        ", xid "LPU64"\n",
2374                        mds->mds_last_transno, obd->obd_last_committed,
2375                        req->rq_xid);
2376         }
2377  out:
2378
2379         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LAST_REPLAY) {
2380                 if (obd && obd->obd_recovering) {
2381                         DEBUG_REQ(D_HA, req, "LAST_REPLAY, queuing reply");
2382                         return target_queue_final_reply(req, rc);
2383                 }
2384                 /* Lost a race with recovery; let the error path DTRT. */
2385                 rc = req->rq_status = -ENOTCONN;
2386         }
2387
2388         target_send_reply(req, rc, fail);
2389         return 0;
2390 }
2391
2392 /* Update the server data on disk.  This stores the new mount_count and also the
2393  * last_rcvd value to disk.  If we don't have a clean shutdown, then the server
2394  * last_rcvd value may be less than that of the clients.  This will alert us
2395  * that we may need to do client recovery.
2396  *
2397  * Also assumes for mds_last_transno that we are not modifying it (no locking).
2398  */
2399 int mds_update_server_data(struct obd_device *obd, int force_sync)
2400 {
2401         struct mds_obd *mds = &obd->u.mds;
2402         struct mds_server_data *msd = mds->mds_server_data;
2403         struct file *filp = mds->mds_rcvd_filp;
2404         struct lvfs_run_ctxt saved;
2405         loff_t off = 0;
2406         int rc;
2407         ENTRY;
2408
2409         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2410         msd->msd_last_transno = cpu_to_le64(mds->mds_last_transno);
2411
2412         CDEBUG(D_SUPER, "MDS mount_count is "LPU64", last_transno is "LPU64"\n",
2413                mds->mds_mount_count, mds->mds_last_transno);
2414         rc = fsfilt_write_record(obd, filp, msd, sizeof(*msd), &off, force_sync);
2415         if (rc)
2416                 CERROR("error writing MDS server data: rc = %d\n", rc);
2417         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2418
2419         RETURN(rc);
2420 }
2421
2422 /* saves last allocated fid counter to file. */
2423 int mds_update_last_fid(struct obd_device *obd, void *handle,
2424                         int force_sync)
2425 {
2426         struct mds_obd *mds = &obd->u.mds;
2427         struct file *filp = mds->mds_fid_filp;
2428         struct lvfs_run_ctxt saved;
2429         loff_t off = 0;
2430         int rc = 0;
2431         ENTRY;
2432
2433         down(&mds->mds_last_fid_sem);
2434         if (mds->mds_last_fid_changed) {
2435                 CDEBUG(D_SUPER, "MDS last_fid is #"LPU64"\n",
2436                        mds->mds_last_fid);
2437
2438                 if (handle) {
2439                         fsfilt_add_journal_cb(obd, mds->mds_sb,
2440                                               mds->mds_last_fid, handle,
2441                                               mds_commit_last_fid_cb, NULL);
2442                 }
2443                 
2444                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2445                 rc = fsfilt_write_record(obd, filp, &mds->mds_last_fid,
2446                                          sizeof(mds->mds_last_fid),
2447                                          &off, force_sync);
2448                 if (rc) {
2449                         CERROR("error writing MDS last_fid #"LPU64
2450                                ", err = %d\n", mds->mds_last_fid, rc);
2451                 } else {
2452                         mds->mds_last_fid_changed = 0;
2453                 }
2454                 
2455                 CDEBUG(D_SUPER, "wrote fid #"LPU64" at idx "
2456                        "%llu: err = %d\n", mds->mds_last_fid,
2457                        off, rc);
2458                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2459         }
2460         up(&mds->mds_last_fid_sem);
2461
2462         RETURN(rc);
2463 }
2464
2465 void mds_set_last_fid(struct obd_device *obd, __u64 fid)
2466 {
2467         struct mds_obd *mds = &obd->u.mds;
2468
2469         down(&mds->mds_last_fid_sem);
2470         if (fid > mds->mds_last_fid) {
2471                 mds->mds_last_fid = fid;
2472                 mds->mds_last_fid_changed = 1;
2473         }
2474         up(&mds->mds_last_fid_sem);
2475 }
2476
2477 void mds_commit_last_transno_cb(struct obd_device *obd,
2478                                 __u64 transno, void *data,
2479                                 int error)
2480 {
2481         obd_transno_commit_cb(obd, transno, error);
2482 }
2483
2484 void mds_commit_last_fid_cb(struct obd_device *obd,
2485                             __u64 fid, void *data,
2486                             int error)
2487 {
2488         if (error) {
2489                 CERROR("%s: fid "LPD64" commit error: %d\n",
2490                        obd->obd_name, fid, error);
2491                 return;
2492         }
2493         
2494         CDEBUG(D_HA, "%s: fid "LPD64" committed\n",
2495                obd->obd_name, fid);
2496 }
2497
2498 /*
2499  * allocates new lustre_id on passed @inode and saves it to inode EA.
2500  */
2501 int mds_alloc_inode_sid(struct obd_device *obd, struct inode *inode,
2502                         void *handle, struct lustre_id *id)
2503 {
2504         struct mds_obd *mds = &obd->u.mds;
2505         int alloc = 0, rc = 0;
2506         ENTRY;
2507
2508         LASSERT(obd != NULL);
2509         LASSERT(inode != NULL);
2510
2511         if (id == NULL) {
2512                 OBD_ALLOC(id, sizeof(*id));
2513                 if (id == NULL)
2514                         RETURN(-ENOMEM);
2515                 alloc = 1;
2516         }
2517
2518         id_group(id) = mds->mds_num;
2519         
2520         down(&mds->mds_last_fid_sem);
2521         mds->mds_last_fid_changed = 1;
2522         id_fid(id) = ++mds->mds_last_fid;
2523         up(&mds->mds_last_fid_sem);
2524
2525         id_ino(id) = inode->i_ino;
2526         id_gen(id) = inode->i_generation;
2527         id_type(id) = (S_IFMT & inode->i_mode);
2528
2529         rc = mds_update_inode_sid(obd, inode, handle, id);
2530         if (rc) {
2531                 CERROR("Can't update inode FID EA, "
2532                        "rc = %d\n", rc);
2533         }
2534
2535         if (alloc)
2536                 OBD_FREE(id, sizeof(*id));
2537         RETURN(rc);
2538 }
2539
2540 /*
2541  * reads inode self id from inode EA. Probably later this should be replaced by
2542  * caching inode self id to avoid raeding it every time it is needed.
2543  */
2544 int mds_read_inode_sid(struct obd_device *obd, struct inode *inode,
2545                        struct lustre_id *id)
2546 {
2547         int rc;
2548         ENTRY;
2549
2550         LASSERT(id != NULL);
2551         LASSERT(obd != NULL);
2552         LASSERT(inode != NULL);
2553
2554         rc = fsfilt_get_sid(obd, inode, &id->li_fid,
2555                             sizeof(id->li_fid));
2556         if (rc < 0) {
2557                 CERROR("fsfilt_get_sid() failed, "
2558                        "rc = %d\n", rc);
2559                 RETURN(rc);
2560         } else if (!rc) {
2561                 rc = -ENODATA;
2562                 RETURN(rc);
2563         } else {
2564                 rc = 0;
2565         }
2566
2567         RETURN(rc);
2568 }
2569
2570 /* updates inode self id in EA. */
2571 int mds_update_inode_sid(struct obd_device *obd, struct inode *inode,
2572                          void *handle, struct lustre_id *id)
2573 {
2574         int rc = 0;
2575         ENTRY;
2576
2577         LASSERT(id != NULL);
2578         LASSERT(obd != NULL);
2579         LASSERT(inode != NULL);
2580         
2581         rc = fsfilt_set_sid(obd, inode, handle, &id->li_fid,
2582                             sizeof(id->li_fid));
2583         if (rc) {
2584                 CERROR("fsfilt_set_sid() failed, rc = %d\n", rc);
2585                 RETURN(rc);
2586         }
2587
2588         RETURN(rc);
2589 }
2590
2591 /* 
2592  * reads inode id on master MDS. This is usualy done by CMOBD to update requests
2593  * to master MDS by correct store cookie, needed to find inode on master MDS
2594  * quickly.
2595  */
2596 int mds_read_inode_mid(struct obd_device *obd, struct inode *inode,
2597                        struct lustre_id *id)
2598 {
2599         int rc;
2600         ENTRY;
2601
2602         LASSERT(id != NULL);
2603         LASSERT(obd != NULL);
2604         LASSERT(inode != NULL);
2605
2606         rc = fsfilt_get_mid(obd, inode, id, sizeof(*id));
2607         if (rc < 0) {
2608                 CERROR("fsfilt_get_mid() failed, "
2609                        "rc = %d\n", rc);
2610                 RETURN(rc);
2611         } else if (!rc) {
2612                 rc = -ENODATA;
2613                 RETURN(rc);
2614         } else {
2615                 rc = 0;
2616         }
2617
2618         RETURN(rc);
2619 }
2620
2621 /*
2622  * updates master inode id. Usualy this is done by CMOBD after an inode is
2623  * created and relationship between cache MDS and master one should be
2624  * established.
2625  */
2626 int mds_update_inode_mid(struct obd_device *obd, struct inode *inode,
2627                          void *handle, struct lustre_id *id)
2628 {
2629         int rc = 0;
2630         ENTRY;
2631
2632         LASSERT(id != NULL);
2633         LASSERT(obd != NULL);
2634         LASSERT(inode != NULL);
2635         
2636         rc = fsfilt_set_mid(obd, inode, handle, id, sizeof(*id));
2637         if (rc) {
2638                 CERROR("fsfilt_set_mid() failed, rc = %d\n", rc);
2639                 RETURN(rc);
2640         }
2641
2642         RETURN(rc);
2643 }
2644
2645 /* mount the file system (secretly) */
2646 static int mds_setup(struct obd_device *obd, obd_count len, void *buf)
2647 {
2648         struct lustre_cfg* lcfg = buf;
2649         struct mds_obd *mds = &obd->u.mds;
2650         char *options = NULL;
2651         struct vfsmount *mnt;
2652         char ns_name[48];
2653         unsigned long page;
2654         int rc = 0;
2655         ENTRY;
2656
2657         dev_clear_rdonly(2);
2658
2659         if (!lcfg->lcfg_inlbuf1 || !lcfg->lcfg_inlbuf2)
2660                 RETURN(rc = -EINVAL);
2661
2662         obd->obd_fsops = fsfilt_get_ops(lcfg->lcfg_inlbuf2);
2663         if (IS_ERR(obd->obd_fsops))
2664                 RETURN(rc = PTR_ERR(obd->obd_fsops));
2665
2666         mds->mds_max_mdsize = sizeof(struct lov_mds_md);
2667
2668         page = __get_free_page(GFP_KERNEL);
2669         if (!page)
2670                 RETURN(-ENOMEM);
2671
2672         options = (char *)page;
2673         memset(options, 0, PAGE_SIZE);
2674
2675         /*
2676          * here we use "iopen_nopriv" hardcoded, because it affects MDS utility
2677          * and the rest of options are passed by mount options. Probably this
2678          * should be moved to somewhere else like startup scripts or lconf.
2679          */
2680         sprintf(options, "iopen_nopriv");
2681
2682         if (lcfg->lcfg_inllen4 > 0 && lcfg->lcfg_inlbuf4)
2683                 sprintf(options + strlen(options), ",%s",
2684                         lcfg->lcfg_inlbuf4);
2685
2686         /* we have to know mdsnum before touching underlying fs -bzzz */
2687         sema_init(&mds->mds_lmv_sem, 1);
2688         mds->mds_lmv_connected = 0;
2689         mds->mds_lmv_name = NULL;
2690         
2691         if (lcfg->lcfg_inllen5 > 0 && lcfg->lcfg_inlbuf5 && 
2692             strncmp(lcfg->lcfg_inlbuf5, "dumb", lcfg->lcfg_inllen5)) {
2693                 class_uuid_t uuid;
2694
2695                 CDEBUG(D_OTHER, "MDS: %s is master for %s\n",
2696                        obd->obd_name, lcfg->lcfg_inlbuf5);
2697
2698                 generate_random_uuid(uuid);
2699                 class_uuid_unparse(uuid, &mds->mds_lmv_uuid);
2700
2701                 OBD_ALLOC(mds->mds_lmv_name, lcfg->lcfg_inllen5);
2702                 if (mds->mds_lmv_name == NULL) 
2703                         RETURN(rc = -ENOMEM);
2704
2705                 memcpy(mds->mds_lmv_name, lcfg->lcfg_inlbuf5,
2706                        lcfg->lcfg_inllen5);
2707                 
2708                 rc = mds_lmv_connect(obd, mds->mds_lmv_name);
2709                 if (rc) {
2710                         OBD_FREE(mds->mds_lmv_name, lcfg->lcfg_inllen5);
2711                         GOTO(err_ops, rc);
2712                 }
2713         }
2714         
2715         mds->mds_obd_type = MDS_MASTER_OBD;
2716
2717         if (lcfg->lcfg_inllen6 > 0 && lcfg->lcfg_inlbuf6 &&
2718             strncmp(lcfg->lcfg_inlbuf6, "dumb", lcfg->lcfg_inllen6)) {
2719                 if (!memcmp(lcfg->lcfg_inlbuf6, "master", strlen("master"))) {
2720                         mds->mds_obd_type = MDS_MASTER_OBD;
2721                 } else if (!memcmp(lcfg->lcfg_inlbuf6, "cache", strlen("cache"))) {
2722                         mds->mds_obd_type = MDS_CACHE_OBD;
2723                 }     
2724         }
2725
2726         mnt = do_kern_mount(lcfg->lcfg_inlbuf2, 0, lcfg->lcfg_inlbuf1, options);
2727         free_page(page);
2728
2729         if (IS_ERR(mnt)) {
2730                 rc = PTR_ERR(mnt);
2731                 CERROR("do_kern_mount failed: rc = %d\n", rc);
2732                 GOTO(err_ops, rc);
2733         }
2734
2735         CDEBUG(D_SUPER, "%s: mnt = %p\n", lcfg->lcfg_inlbuf1, mnt);
2736
2737         mds->mds_last_fid_changed = 0;
2738         sema_init(&mds->mds_epoch_sem, 1);
2739         sema_init(&mds->mds_last_fid_sem, 1);
2740         atomic_set(&mds->mds_real_clients, 0);
2741         spin_lock_init(&mds->mds_transno_lock);
2742         sema_init(&mds->mds_orphan_recovery_sem, 1);
2743         mds->mds_max_cookiesize = sizeof(struct llog_cookie);
2744
2745         sprintf(ns_name, "mds-%s", obd->obd_uuid.uuid);
2746         obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER);
2747
2748         if (obd->obd_namespace == NULL) {
2749                 mds_cleanup(obd, 0);
2750                 GOTO(err_put, rc = -ENOMEM);
2751         }
2752         ldlm_register_intent(obd->obd_namespace, mds_intent_policy);
2753
2754         rc = mds_fs_setup(obd, mnt);
2755         if (rc) {
2756                 CERROR("%s: MDS filesystem method init failed: rc = %d\n",
2757                        obd->obd_name, rc);
2758                 GOTO(err_ns, rc);
2759         }
2760         
2761         rc = llog_start_commit_thread();
2762         if (rc < 0)
2763                 GOTO(err_fs, rc);
2764
2765         if (lcfg->lcfg_inllen3 > 0 && lcfg->lcfg_inlbuf3 &&
2766             strncmp(lcfg->lcfg_inlbuf3, "dumb", lcfg->lcfg_inllen3)) {
2767                 class_uuid_t uuid;
2768
2769                 generate_random_uuid(uuid);
2770                 class_uuid_unparse(uuid, &mds->mds_lov_uuid);
2771
2772                 OBD_ALLOC(mds->mds_profile, lcfg->lcfg_inllen3);
2773                 if (mds->mds_profile == NULL)
2774                         GOTO(err_fs, rc = -ENOMEM);
2775
2776                 memcpy(mds->mds_profile, lcfg->lcfg_inlbuf3,
2777                        lcfg->lcfg_inllen3);
2778         }
2779
2780         /* 
2781          * setup root dir and files ID dir if lmv already connected, or there is
2782          * not lmv at all.
2783          */
2784         if (mds->mds_lmv_exp || (lcfg->lcfg_inllen3 > 0 && lcfg->lcfg_inlbuf3 &&
2785                                  strncmp(lcfg->lcfg_inlbuf3, "dumb", lcfg->lcfg_inllen3)))
2786         {
2787                 rc = mds_fs_setup_rootid(obd);
2788                 if (rc)
2789                         GOTO(err_fs, rc);
2790
2791                 rc = mds_fs_setup_virtid(obd);
2792                 if (rc)
2793                         GOTO(err_fs, rc);
2794         }
2795
2796         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
2797                            "mds_ldlm_client", &obd->obd_ldlm_client);
2798         obd->obd_replayable = 1;
2799
2800         rc = mds_postsetup(obd);
2801         if (rc)
2802                 GOTO(err_fs, rc);
2803
2804         RETURN(0);
2805
2806 err_fs:
2807         /* No extra cleanup needed for llog_init_commit_thread() */
2808         mds_fs_cleanup(obd, 0);
2809 err_ns:
2810         ldlm_namespace_free(obd->obd_namespace, 0);
2811         obd->obd_namespace = NULL;
2812 err_put:
2813         unlock_kernel();
2814         mntput(mds->mds_vfsmnt);
2815         mds->mds_sb = 0;
2816         lock_kernel();
2817 err_ops:
2818         fsfilt_put_ops(obd->obd_fsops);
2819         return rc;
2820 }
2821
2822 static int mds_postsetup(struct obd_device *obd)
2823 {
2824         struct mds_obd *mds = &obd->u.mds;
2825         int rc = 0;
2826         ENTRY;
2827
2828         rc = obd_llog_setup(obd, &obd->obd_llogs, LLOG_CONFIG_ORIG_CTXT, 
2829                             obd, 0, NULL, &llog_lvfs_ops);
2830         if (rc)
2831                 RETURN(rc);
2832
2833         if (mds->mds_profile) {
2834                 struct llog_ctxt *lgctxt;
2835                 struct lvfs_run_ctxt saved;
2836                 struct lustre_profile *lprof;
2837                 struct config_llog_instance cfg;
2838
2839                 cfg.cfg_instance = NULL;
2840                 cfg.cfg_uuid = mds->mds_lov_uuid;
2841                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2842
2843                 lgctxt = llog_get_context(&obd->obd_llogs, LLOG_CONFIG_ORIG_CTXT);
2844                 if (!lgctxt) {
2845                         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2846                         GOTO(err_llog, rc = -EINVAL);
2847                 }
2848                 
2849                 rc = class_config_process_llog(lgctxt, mds->mds_profile, &cfg);
2850                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2851
2852                 if (rc)
2853                         GOTO(err_llog, rc);
2854
2855                 lprof = class_get_profile(mds->mds_profile);
2856                 if (lprof == NULL) {
2857                         CERROR("No profile found: %s\n", mds->mds_profile);
2858                         GOTO(err_cleanup, rc = -ENOENT);
2859                 }
2860                 rc = mds_lov_connect(obd, lprof->lp_lov);
2861                 if (rc)
2862                         GOTO(err_cleanup, rc);
2863
2864                 rc = mds_lmv_postsetup(obd);
2865                 if (rc)
2866                         GOTO(err_cleanup, rc);
2867         }
2868
2869         RETURN(rc);
2870 err_cleanup:
2871         mds_lov_clean(obd);
2872 err_llog:
2873         obd_llog_cleanup(llog_get_context(&obd->obd_llogs,
2874                                           LLOG_CONFIG_ORIG_CTXT));
2875         return rc;
2876 }
2877
2878 int mds_postrecov(struct obd_device *obd)
2879 {
2880         struct mds_obd *mds = &obd->u.mds;
2881         struct llog_ctxt *ctxt;
2882         int rc, item = 0, valsize;
2883          __u32 group;
2884         ENTRY;
2885
2886         LASSERT(!obd->obd_recovering);
2887         ctxt = llog_get_context(&obd->obd_llogs, LLOG_UNLINK_ORIG_CTXT);
2888         LASSERT(ctxt != NULL);
2889
2890         /* set nextid first, so we are sure it happens */
2891         rc = mds_lov_set_nextid(obd);
2892         if (rc) {
2893                 CERROR("%s: mds_lov_set_nextid failed\n", obd->obd_name);
2894                 GOTO(out, rc);
2895         }
2896
2897         /* clean PENDING dir */
2898         rc = mds_cleanup_orphans(obd);
2899         if (rc < 0)
2900                 GOTO(out, rc);
2901         item = rc;
2902
2903         group = FILTER_GROUP_FIRST_MDS + mds->mds_num;
2904         valsize = sizeof(group);
2905         rc = obd_set_info(mds->mds_lov_exp, strlen("mds_conn"), "mds_conn",
2906                           valsize, &group);
2907         if (rc)
2908                 GOTO(out, rc);
2909
2910         rc = llog_connect(ctxt, obd->u.mds.mds_lov_desc.ld_tgt_count,
2911                           NULL, NULL, NULL);
2912         if (rc) {
2913                 CERROR("%s: failed at llog_origin_connect: %d\n", 
2914                        obd->obd_name, rc);
2915                 GOTO(out, rc);
2916         }
2917
2918         /* remove the orphaned precreated objects */
2919         rc = mds_lov_clearorphans(mds, NULL /* all OSTs */);
2920         if (rc)
2921                 GOTO(err_llog, rc);
2922
2923 out:
2924         RETURN(rc < 0 ? rc : item);
2925
2926 err_llog:
2927         /* cleanup all llogging subsystems */
2928         rc = obd_llog_finish(obd, &obd->obd_llogs,
2929                              mds->mds_lov_desc.ld_tgt_count);
2930         if (rc)
2931                 CERROR("%s: failed to cleanup llogging subsystems\n",
2932                         obd->obd_name);
2933         goto out;
2934 }
2935
2936 int mds_lov_clean(struct obd_device *obd)
2937 {
2938         struct mds_obd *mds = &obd->u.mds;
2939         ENTRY;
2940
2941         if (mds->mds_profile) {
2942                 char * cln_prof;
2943                 struct llog_ctxt *llctx;
2944                 struct lvfs_run_ctxt saved;
2945                 struct config_llog_instance cfg;
2946                 int len = strlen(mds->mds_profile) + sizeof("-clean") + 1;
2947
2948                 OBD_ALLOC(cln_prof, len);
2949                 sprintf(cln_prof, "%s-clean", mds->mds_profile);
2950
2951                 cfg.cfg_instance = NULL;
2952                 cfg.cfg_uuid = mds->mds_lov_uuid;
2953
2954                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2955                 llctx = llog_get_context(&obd->obd_llogs,
2956                                          LLOG_CONFIG_ORIG_CTXT);
2957                 class_config_process_llog(llctx, cln_prof, &cfg);
2958                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
2959
2960                 OBD_FREE(cln_prof, len);
2961                 OBD_FREE(mds->mds_profile, strlen(mds->mds_profile) + 1);
2962                 mds->mds_profile = NULL;
2963         }
2964         RETURN(0);
2965 }
2966
2967 int mds_lmv_clean(struct obd_device *obd)
2968 {
2969         struct mds_obd *mds = &obd->u.mds;
2970         ENTRY;
2971
2972         if (mds->mds_lmv_name) {
2973                 OBD_FREE(mds->mds_lmv_name, strlen(mds->mds_lmv_name) + 1);
2974                 mds->mds_lmv_name = NULL;
2975         }
2976         RETURN(0);
2977 }
2978
2979 static int mds_precleanup(struct obd_device *obd, int flags)
2980 {
2981         int rc = 0;
2982         ENTRY;
2983
2984         mds_lmv_clean(obd);
2985         mds_lov_disconnect(obd, flags);
2986         mds_lov_clean(obd);
2987         obd_llog_cleanup(llog_get_context(&obd->obd_llogs, LLOG_CONFIG_ORIG_CTXT));
2988         RETURN(rc);
2989 }
2990
2991 static int mds_cleanup(struct obd_device *obd, int flags)
2992 {
2993         struct mds_obd *mds = &obd->u.mds;
2994         ENTRY;
2995
2996         if (mds->mds_sb == NULL)
2997                 RETURN(0);
2998
2999         mds_update_server_data(obd, 1);
3000         mds_update_last_fid(obd, NULL, 1);
3001         
3002         if (mds->mds_lov_objids != NULL) {
3003                 int size = mds->mds_lov_desc.ld_tgt_count *
3004                         sizeof(obd_id);
3005                 OBD_FREE(mds->mds_lov_objids, size);
3006         }
3007         mds_fs_cleanup(obd, flags);
3008
3009         unlock_kernel();
3010
3011         /*
3012          * 2 seems normal on mds, (may_umount() also expects 2 fwiw), but we
3013          * only see 1 at this point in obdfilter.
3014          */
3015         if (atomic_read(&obd->u.mds.mds_vfsmnt->mnt_count) > 2)
3016                 CERROR("%s: mount busy, mnt_count %d != 2\n", obd->obd_name,
3017                        atomic_read(&obd->u.mds.mds_vfsmnt->mnt_count));
3018
3019         mntput(mds->mds_vfsmnt);
3020         mds->mds_sb = 0;
3021
3022         ldlm_namespace_free(obd->obd_namespace, flags & OBD_OPT_FORCE);
3023
3024         spin_lock_bh(&obd->obd_processing_task_lock);
3025         if (obd->obd_recovering) {
3026                 target_cancel_recovery_timer(obd);
3027                 obd->obd_recovering = 0;
3028         }
3029         spin_unlock_bh(&obd->obd_processing_task_lock);
3030
3031         lock_kernel();
3032         dev_clear_rdonly(2);
3033         fsfilt_put_ops(obd->obd_fsops);
3034
3035         RETURN(0);
3036 }
3037
3038 static void fixup_handle_for_resent_req(struct ptlrpc_request *req,
3039                                         int offset,
3040                                         struct ldlm_lock *new_lock,
3041                                         struct ldlm_lock **old_lock,
3042                                         struct lustre_handle *lockh)
3043 {
3044         struct obd_export *exp = req->rq_export;
3045         struct obd_device *obd = exp->exp_obd;
3046         struct ldlm_request *dlmreq =
3047                 lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*dlmreq));
3048         struct lustre_handle remote_hdl = dlmreq->lock_handle1;
3049         struct list_head *iter;
3050
3051         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
3052                 return;
3053
3054         l_lock(&obd->obd_namespace->ns_lock);
3055         list_for_each(iter, &exp->exp_ldlm_data.led_held_locks) {
3056                 struct ldlm_lock *lock;
3057                 lock = list_entry(iter, struct ldlm_lock, l_export_chain);
3058                 if (lock == new_lock)
3059                         continue;
3060                 if (lock->l_remote_handle.cookie == remote_hdl.cookie) {
3061                         lockh->cookie = lock->l_handle.h_cookie;
3062                         LDLM_DEBUG(lock, "restoring lock cookie");
3063                         DEBUG_REQ(D_HA, req, "restoring lock cookie "LPX64,
3064                                   lockh->cookie);
3065                         if (old_lock)
3066                                 *old_lock = LDLM_LOCK_GET(lock);
3067                         l_unlock(&obd->obd_namespace->ns_lock);
3068                         return;
3069                 }
3070         }
3071         l_unlock(&obd->obd_namespace->ns_lock);
3072
3073         /* If the xid matches, then we know this is a resent request,
3074          * and allow it. (It's probably an OPEN, for which we don't
3075          * send a lock */
3076         if (req->rq_xid == exp->exp_mds_data.med_mcd->mcd_last_xid)
3077                 return;
3078
3079         /* This remote handle isn't enqueued, so we never received or
3080          * processed this request.  Clear MSG_RESENT, because it can
3081          * be handled like any normal request now. */
3082
3083         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
3084
3085         DEBUG_REQ(D_HA, req, "no existing lock with rhandle "LPX64,
3086                   remote_hdl.cookie);
3087 }
3088
3089 int intent_disposition(struct ldlm_reply *rep, int flag)
3090 {
3091         if (!rep)
3092                 return 0;
3093         return (rep->lock_policy_res1 & flag);
3094 }
3095
3096 void intent_set_disposition(struct ldlm_reply *rep, int flag)
3097 {
3098         if (!rep)
3099                 return;
3100         rep->lock_policy_res1 |= flag;
3101 }
3102
3103 static int mds_intent_policy(struct ldlm_namespace *ns,
3104                              struct ldlm_lock **lockp, void *req_cookie,
3105                              ldlm_mode_t mode, int flags, void *data)
3106 {
3107         struct ptlrpc_request *req = req_cookie;
3108         struct ldlm_lock *lock = *lockp;
3109         struct ldlm_intent *it;
3110         struct mds_obd *mds = &req->rq_export->exp_obd->u.mds;
3111         struct ldlm_reply *rep;
3112         struct lustre_handle lockh[2] = {{0}, {0}};
3113         struct ldlm_lock *new_lock = NULL;
3114         int getattr_part = MDS_INODELOCK_UPDATE;
3115         int rc, repsize[4] = { sizeof(struct ldlm_reply),
3116                                sizeof(struct mds_body),
3117                                mds->mds_max_mdsize,
3118                                mds->mds_max_cookiesize };
3119         int offset = MDS_REQ_INTENT_REC_OFF; 
3120         ENTRY;
3121
3122         LASSERT(req != NULL);
3123         MD_COUNTER_INCREMENT(req->rq_export->exp_obd, intent_lock);
3124
3125         if (req->rq_reqmsg->bufcount <= MDS_REQ_INTENT_IT_OFF) {
3126                 /* No intent was provided */
3127                 int size = sizeof(struct ldlm_reply);
3128                 rc = lustre_pack_reply(req, 1, &size, NULL);
3129                 LASSERT(rc == 0);
3130                 RETURN(0);
3131         }
3132
3133         it = lustre_swab_reqbuf(req, MDS_REQ_INTENT_IT_OFF, sizeof(*it),
3134                                 lustre_swab_ldlm_intent);
3135         if (it == NULL) {
3136                 CERROR("Intent missing\n");
3137                 RETURN(req->rq_status = -EFAULT);
3138         }
3139
3140         LDLM_DEBUG(lock, "intent policy, opc: %s", ldlm_it2str(it->opc));
3141
3142         rc = lustre_pack_reply(req, 3, repsize, NULL);
3143         if (rc)
3144                 RETURN(req->rq_status = rc);
3145
3146         rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*rep));
3147         LASSERT(rep != NULL);
3148
3149         intent_set_disposition(rep, DISP_IT_EXECD);
3150
3151         /* execute policy */
3152         switch ((long)it->opc) {
3153         case IT_OPEN:
3154         case IT_CREAT|IT_OPEN:
3155                 /* XXX swab here to assert that an mds_open reint
3156                  * packet is following */
3157                 fixup_handle_for_resent_req(req, MDS_REQ_INTENT_LOCKREQ_OFF, 
3158                                             lock, NULL, lockh);
3159                 rep->lock_policy_res2 = mds_reint(req, offset, lockh);
3160 #if 0
3161                 /* We abort the lock if the lookup was negative and
3162                  * we did not make it to the OPEN portion */
3163                 if (!intent_disposition(rep, DISP_LOOKUP_EXECD))
3164                         RETURN(ELDLM_LOCK_ABORTED);
3165                 if (intent_disposition(rep, DISP_LOOKUP_NEG) &&
3166                     !intent_disposition(rep, DISP_OPEN_OPEN))
3167 #endif
3168                 /* IT_OPEN may return lock on cross-node dentry
3169                  * that we want to hold during attr retrival -bzzz */
3170                 if (rc != 0 || lockh[0].cookie == 0)
3171                         RETURN(ELDLM_LOCK_ABORTED);
3172                 break;
3173         case IT_LOOKUP:
3174                 getattr_part = MDS_INODELOCK_LOOKUP;
3175         case IT_CHDIR:
3176         case IT_GETATTR:
3177                 getattr_part |= MDS_INODELOCK_LOOKUP;
3178         case IT_READDIR:
3179                 fixup_handle_for_resent_req(req, MDS_REQ_INTENT_LOCKREQ_OFF, 
3180                                             lock, &new_lock, lockh);
3181                 rep->lock_policy_res2 = mds_getattr_lock(req, offset, lockh,
3182                                                          getattr_part);
3183                 /* FIXME: LDLM can set req->rq_status. MDS sets
3184                    policy_res{1,2} with disposition and status.
3185                    - replay: returns 0 & req->status is old status
3186                    - otherwise: returns req->status */
3187                 if (intent_disposition(rep, DISP_LOOKUP_NEG))
3188                         rep->lock_policy_res2 = 0;
3189                 if (!intent_disposition(rep, DISP_LOOKUP_POS) ||
3190                     rep->lock_policy_res2)
3191                         RETURN(ELDLM_LOCK_ABORTED);
3192                 if (req->rq_status != 0) {
3193                         LBUG();
3194                         rep->lock_policy_res2 = req->rq_status;
3195                         RETURN(ELDLM_LOCK_ABORTED);
3196                 }
3197                 break;
3198         case IT_UNLINK:
3199                 rc = mds_lock_and_check_slave(offset, req, lockh);
3200                 if ((rep->lock_policy_res2 = rc)) {
3201                         if (rc == ENOLCK)
3202                                 rep->lock_policy_res2 = 0;
3203                         RETURN(ELDLM_LOCK_ABORTED);
3204                 }
3205                 break;
3206         default:
3207                 CERROR("Unhandled intent "LPD64"\n", it->opc);
3208                 LBUG();
3209         }
3210
3211         /* By this point, whatever function we called above must have either
3212          * filled in 'lockh', been an intent replay, or returned an error.  We
3213          * want to allow replayed RPCs to not get a lock, since we would just
3214          * drop it below anyways because lock replay is done separately by the
3215          * client afterwards.  For regular RPCs we want to give the new lock to
3216          * the client instead of whatever lock it was about to get. */
3217         if (new_lock == NULL)
3218                 new_lock = ldlm_handle2lock(&lockh[0]);
3219         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY))
3220                 RETURN(0);
3221
3222         LASSERTF(new_lock != NULL, "op "LPX64" lockh "LPX64"\n",
3223                  it->opc, lockh[0].cookie);
3224  
3225         /* If we've already given this lock to a client once, then we should
3226          * have no readers or writers.  Otherwise, we should have one reader
3227          * _or_ writer ref (which will be zeroed below) before returning the
3228          * lock to a client. */
3229         if (new_lock->l_export == req->rq_export) {
3230                 LASSERT(new_lock->l_readers + new_lock->l_writers == 0);
3231         } else {
3232                 LASSERT(new_lock->l_export == NULL);
3233                 LASSERT(new_lock->l_readers + new_lock->l_writers == 1);
3234         }
3235
3236         *lockp = new_lock;
3237
3238         if (new_lock->l_export == req->rq_export) {
3239                 /* Already gave this to the client, which means that we
3240                  * reconstructed a reply. */
3241                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
3242                         MSG_RESENT);
3243                 RETURN(ELDLM_LOCK_REPLACED);
3244         }
3245
3246         /* Fixup the lock to be given to the client */
3247         l_lock(&new_lock->l_resource->lr_namespace->ns_lock);
3248         new_lock->l_readers = 0;
3249         new_lock->l_writers = 0;
3250
3251         new_lock->l_export = class_export_get(req->rq_export);
3252         list_add(&new_lock->l_export_chain,
3253                  &new_lock->l_export->exp_ldlm_data.led_held_locks);
3254
3255         new_lock->l_blocking_ast = lock->l_blocking_ast;
3256         new_lock->l_completion_ast = lock->l_completion_ast;
3257
3258         memcpy(&new_lock->l_remote_handle, &lock->l_remote_handle,
3259                sizeof(lock->l_remote_handle));
3260
3261         new_lock->l_flags &= ~LDLM_FL_LOCAL;
3262
3263         LDLM_LOCK_PUT(new_lock);
3264         l_unlock(&new_lock->l_resource->lr_namespace->ns_lock);
3265
3266         RETURN(ELDLM_LOCK_REPLACED);
3267 }
3268
3269 int mds_attach(struct obd_device *dev, obd_count len, void *data)
3270 {
3271         struct lprocfs_static_vars lvars;
3272         int rc = 0;
3273
3274         lprocfs_init_multi_vars(0, &lvars);
3275
3276         rc = lprocfs_obd_attach(dev, lvars.obd_vars);
3277         if (rc)
3278                 return rc;
3279
3280         return lprocfs_alloc_md_stats(dev, 0);
3281 }
3282
3283 int mds_detach(struct obd_device *dev)
3284 {
3285         lprocfs_free_md_stats(dev);
3286         return lprocfs_obd_detach(dev);
3287 }
3288
3289 int mdt_attach(struct obd_device *dev, obd_count len, void *data)
3290 {
3291         struct lprocfs_static_vars lvars;
3292
3293         lprocfs_init_multi_vars(1, &lvars);
3294         return lprocfs_obd_attach(dev, lvars.obd_vars);
3295 }
3296
3297 int mdt_detach(struct obd_device *dev)
3298 {
3299         return lprocfs_obd_detach(dev);
3300 }
3301
3302 static int mdt_setup(struct obd_device *obd, obd_count len, void *buf)
3303 {
3304         struct mds_obd *mds = &obd->u.mds;
3305         int rc = 0;
3306         ENTRY;
3307
3308         mds->mds_service =
3309                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
3310                                 MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL,
3311                                 mds_handle, "mds", obd->obd_proc_entry);
3312
3313         if (!mds->mds_service) {
3314                 CERROR("failed to start service\n");
3315                 RETURN(-ENOMEM);
3316         }
3317
3318         rc = ptlrpc_start_n_threads(obd, mds->mds_service, MDT_NUM_THREADS,
3319                                     "ll_mdt");
3320         if (rc)
3321                 GOTO(err_thread, rc);
3322
3323         mds->mds_setattr_service =
3324                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
3325                                 MDS_SETATTR_PORTAL, MDC_REPLY_PORTAL,
3326                                 mds_handle, "mds_setattr",
3327                                 obd->obd_proc_entry);
3328         if (!mds->mds_setattr_service) {
3329                 CERROR("failed to start getattr service\n");
3330                 GOTO(err_thread, rc = -ENOMEM);
3331         }
3332
3333         rc = ptlrpc_start_n_threads(obd, mds->mds_setattr_service,
3334                                     MDT_NUM_THREADS, "ll_mdt_attr");
3335         if (rc)
3336                 GOTO(err_thread2, rc);
3337
3338         mds->mds_readpage_service =
3339                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
3340                                 MDS_READPAGE_PORTAL, MDC_REPLY_PORTAL,
3341                                 mds_handle, "mds_readpage",
3342                                 obd->obd_proc_entry);
3343         if (!mds->mds_readpage_service) {
3344                 CERROR("failed to start readpage service\n");
3345                 GOTO(err_thread2, rc = -ENOMEM);
3346         }
3347
3348         rc = ptlrpc_start_n_threads(obd, mds->mds_readpage_service,
3349                                     MDT_NUM_THREADS, "ll_mdt_rdpg");
3350
3351         if (rc)
3352                 GOTO(err_thread3, rc);
3353
3354         RETURN(0);
3355
3356 err_thread3:
3357         ptlrpc_unregister_service(mds->mds_readpage_service);
3358 err_thread2:
3359         ptlrpc_unregister_service(mds->mds_setattr_service);
3360 err_thread:
3361         ptlrpc_unregister_service(mds->mds_service);
3362         return rc;
3363 }
3364
3365 static int mdt_cleanup(struct obd_device *obd, int flags)
3366 {
3367         struct mds_obd *mds = &obd->u.mds;
3368         ENTRY;
3369
3370         ptlrpc_stop_all_threads(mds->mds_readpage_service);
3371         ptlrpc_unregister_service(mds->mds_readpage_service);
3372
3373         ptlrpc_stop_all_threads(mds->mds_setattr_service);
3374         ptlrpc_unregister_service(mds->mds_setattr_service);
3375
3376         ptlrpc_stop_all_threads(mds->mds_service);
3377         ptlrpc_unregister_service(mds->mds_service);
3378
3379         RETURN(0);
3380 }
3381
3382 static struct dentry *mds_lvfs_id2dentry(__u64 ino, __u32 gen,
3383                                          __u64 gr, void *data)
3384 {
3385         struct lustre_id id;
3386         struct obd_device *obd = data;
3387         
3388         id_ino(&id) = ino;
3389         id_gen(&id) = gen;
3390         return mds_id2dentry(obd, &id, NULL);
3391 }
3392
3393 static int mds_get_info(struct obd_export *exp, __u32 keylen,
3394                         void *key, __u32 *valsize, void *val)
3395 {
3396         struct obd_device *obd;
3397         struct mds_obd *mds;
3398         ENTRY;
3399
3400         obd = class_exp2obd(exp);
3401         mds = &obd->u.mds;
3402         
3403         if (obd == NULL) {
3404                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
3405                        exp->exp_handle.h_cookie);
3406                 RETURN(-EINVAL);
3407         }
3408
3409         if (keylen >= strlen("reint_log") && memcmp(key, "reint_log", 9) == 0) {
3410                 /* get log_context handle. */
3411                 unsigned long *llh_handle = val;
3412                 *valsize = sizeof(unsigned long);
3413                 *llh_handle = (unsigned long)obd->obd_llog_ctxt[LLOG_REINT_ORIG_CTXT];
3414                 RETURN(0);
3415         }
3416         if (keylen >= strlen("cache_sb") && memcmp(key, "cache_sb", 8) == 0) {
3417                 /* get log_context handle. */
3418                 unsigned long *sb = val;
3419                 *valsize = sizeof(unsigned long);
3420                 *sb = (unsigned long)obd->u.mds.mds_sb;
3421                 RETURN(0);
3422         }
3423
3424         if (keylen >= strlen("mdsize") && memcmp(key, "mdsize", keylen) == 0) {
3425                 __u32 *mdsize = val;
3426                 *valsize = sizeof(*mdsize);
3427                 *mdsize = mds->mds_max_mdsize;
3428                 RETURN(0);
3429         }
3430
3431         if (keylen >= strlen("mdsnum") && strcmp(key, "mdsnum") == 0) {
3432                 __u32 *mdsnum = val;
3433                 *valsize = sizeof(*mdsnum);
3434                 *mdsnum = mds->mds_num;
3435                 RETURN(0);
3436         }
3437
3438         if (keylen >= strlen("rootid") && strcmp(key, "rootid") == 0) {
3439                 struct lustre_id *rootid = val;
3440                 *valsize = sizeof(struct lustre_id);
3441                 *rootid = mds->mds_rootid;
3442                 RETURN(0);
3443         }
3444
3445         CDEBUG(D_IOCTL, "invalid key\n");
3446         RETURN(-EINVAL);
3447
3448 }
3449 struct lvfs_callback_ops mds_lvfs_ops = {
3450         l_id2dentry:     mds_lvfs_id2dentry,
3451 };
3452
3453 int mds_preprw(int cmd, struct obd_export *exp, struct obdo *oa,
3454                 int objcount, struct obd_ioobj *obj,
3455                 int niocount, struct niobuf_remote *nb,
3456                 struct niobuf_local *res,
3457                 struct obd_trans_info *oti);
3458
3459 int mds_commitrw(int cmd, struct obd_export *exp, struct obdo *oa,
3460                  int objcount, struct obd_ioobj *obj, int niocount,
3461                  struct niobuf_local *res, struct obd_trans_info *oti,
3462                  int rc);
3463
3464 /* use obd ops to offer management infrastructure */
3465 static struct obd_ops mds_obd_ops = {
3466         .o_owner           = THIS_MODULE,
3467         .o_attach          = mds_attach,
3468         .o_detach          = mds_detach,
3469         .o_connect         = mds_connect,
3470         .o_connect_post    = mds_connect_post,
3471         .o_init_export     = mds_init_export,
3472         .o_destroy_export  = mds_destroy_export,
3473         .o_disconnect      = mds_disconnect,
3474         .o_setup           = mds_setup,
3475         .o_precleanup      = mds_precleanup,
3476         .o_cleanup         = mds_cleanup,
3477         .o_postrecov       = mds_postrecov,
3478         .o_statfs          = mds_obd_statfs,
3479         .o_iocontrol       = mds_iocontrol,
3480         .o_create          = mds_obd_create,
3481         .o_destroy         = mds_obd_destroy,
3482         .o_llog_init       = mds_llog_init,
3483         .o_llog_finish     = mds_llog_finish,
3484         .o_notify          = mds_notify,
3485         .o_get_info        = mds_get_info,
3486         .o_set_info        = mds_set_info,
3487         .o_preprw          = mds_preprw, 
3488         .o_commitrw        = mds_commitrw,
3489 };
3490
3491 static struct obd_ops mdt_obd_ops = {
3492         .o_owner           = THIS_MODULE,
3493         .o_attach          = mdt_attach,
3494         .o_detach          = mdt_detach,
3495         .o_setup           = mdt_setup,
3496         .o_cleanup         = mdt_cleanup,
3497 };
3498
3499 static int __init mds_init(void)
3500 {
3501         struct lprocfs_static_vars lvars;
3502
3503         mds_group_hash_init();
3504
3505         lprocfs_init_multi_vars(0, &lvars);
3506         class_register_type(&mds_obd_ops, NULL, lvars.module_vars,
3507                             LUSTRE_MDS_NAME);
3508         lprocfs_init_multi_vars(1, &lvars);
3509         class_register_type(&mdt_obd_ops, NULL, lvars.module_vars,
3510                             LUSTRE_MDT_NAME);
3511
3512         return 0;
3513 }
3514
3515 static void /*__exit*/ mds_exit(void)
3516 {
3517         mds_group_hash_cleanup();
3518
3519         class_unregister_type(LUSTRE_MDS_NAME);
3520         class_unregister_type(LUSTRE_MDT_NAME);
3521 }
3522
3523 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
3524 MODULE_DESCRIPTION("Lustre Metadata Server (MDS)");
3525 MODULE_LICENSE("GPL");
3526
3527 module_init(mds_init);
3528 module_exit(mds_exit);