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