1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
5 * Lustre Metadata Server (mds) request handler
7 * Copyright (c) 2001, 2002 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>
13 * This file is part of Lustre, http://www.lustre.org.
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.
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.
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.
30 #define DEBUG_SUBSYSTEM S_MDS
32 #include <linux/module.h>
33 #include <linux/lustre_mds.h>
34 #include <linux/lustre_dlm.h>
35 #include <linux/init.h>
36 #include <linux/obd_class.h>
37 #include <linux/random.h>
38 #include <linux/locks.h>
39 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
40 #include <linux/buffer_head.h>
41 #include <linux/workqueue.h>
43 #include <linux/obd_lov.h>
44 #include <linux/lustre_mds.h>
45 #include <linux/lustre_fsfilt.h>
46 #include <linux/lprocfs_status.h>
48 extern kmem_cache_t *mds_file_cache;
49 extern inline struct mds_obd *mds_req2mds(struct ptlrpc_request *req);
50 extern void mds_start_transno(struct mds_obd *mds);
51 extern int mds_finish_transno(struct mds_obd *mds, void *handle,
52 struct ptlrpc_request *req, int rc);
53 extern int enqueue_ordered_locks(int lock_mode, struct obd_device *obd,
54 struct ldlm_res_id *p1_res_id,
55 struct ldlm_res_id *p2_res_id,
56 struct ldlm_res_id *c1_res_id,
57 struct ldlm_res_id *c2_res_id,
58 struct lustre_handle *p1_lockh,
59 struct lustre_handle *p2_lockh,
60 struct lustre_handle *c1_lockh,
61 struct lustre_handle *c2_lockh);
63 int mds_open(struct mds_update_record *rec, int offset,
64 struct ptlrpc_request *req, struct lustre_handle *child_lockh)
66 struct mds_obd *mds = mds_req2mds(req);
67 struct obd_device *obd = req->rq_export->exp_obd;
68 struct ldlm_reply *rep = lustre_msg_buf(req->rq_repmsg, 0);
70 struct mds_body *body = lustre_msg_buf(req->rq_repmsg, 1);
71 struct dentry *dchild, *parent;
72 struct mds_export_data *med;
73 struct mds_file_data *mfd = NULL;
74 struct ldlm_res_id child_res_id = { .name = {0} };
75 struct lustre_handle parent_lockh;
76 int rc = 0, parent_mode, child_mode = LCK_PR, lock_flags, created = 0;
79 #warning replay of open needs to be redone
80 /* was this animal open already and the client lost the reply? */
81 /* XXX need some way to detect a reopen, to avoid locked list walks */
82 med = &req->rq_export->exp_mds_data;
84 spin_lock(&med->med_open_lock);
85 list_for_each(tmp, &med->med_open_head) {
86 mfd = list_entry(tmp, typeof(*mfd), mfd_list);
87 if (!memcmp(&mfd->mfd_clienthandle, &body->handle,
88 sizeof(mfd->mfd_clienthandle)) &&
89 body->fid1.id == mfd->mfd_file->f_dentry->d_inode->i_ino) {
90 dchild = mfd->mfd_file->f_dentry;
91 spin_unlock(&med->med_open_lock);
92 CERROR("Re opening "LPD64"\n", body->fid1.id);
93 GOTO(out_pack, rc = 0);
96 spin_unlock(&med->med_open_lock);
98 rep->lock_policy_res1 |= IT_OPEN_LOOKUP;
99 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
100 CERROR("test case OBD_FAIL_MDS_OPEN_PACK\n");
101 req->rq_status = -ENOMEM;
105 /* Step 1: Find and lock the parent */
106 parent_mode = (rec->ur_flags & O_CREAT) ? LCK_PW : LCK_PR;
107 parent = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, parent_mode,
109 if (IS_ERR(parent)) {
110 rc = PTR_ERR(parent);
111 CERROR("parent lookup error %d\n", rc);
115 LASSERT(parent->d_inode);
117 /* Step 2: Lookup the child */
118 dchild = lookup_one_len(lustre_msg_buf(req->rq_reqmsg, 3),
119 parent, req->rq_reqmsg->buflens[3] - 1);
121 GOTO(out_step_2, rc = PTR_ERR(dchild));
124 rep->lock_policy_res1 |= IT_OPEN_POS;
126 rep->lock_policy_res1 |= IT_OPEN_NEG;
128 /* Step 3: If the child was negative, and we're supposed to,
130 if ((rec->ur_flags & O_CREAT) && !dchild->d_inode) {
133 mds_start_transno(mds);
134 rep->lock_policy_res1 |= IT_OPEN_CREATE;
135 handle = fsfilt_start(obd, parent->d_inode, FSFILT_OP_CREATE);
136 if (IS_ERR(handle)) {
137 rc = PTR_ERR(handle);
138 mds_finish_transno(mds, handle, req, rc);
139 GOTO(out_step_3, rc);
141 rc = vfs_create(parent->d_inode, dchild, rec->ur_mode);
142 rc = mds_finish_transno(mds, handle, req, rc);
143 err = fsfilt_commit(obd, parent->d_inode, handle);
145 CERROR("error on commit: err = %d\n", err);
148 GOTO(out_step_3, rc);
152 } else if (!dchild->d_inode) {
153 /* It's negative and we weren't supposed to create it */
154 GOTO(out_step_3, rc = -ENOENT);
157 /* Step 4: It's positive, so lock the child */
158 child_res_id.name[0] = dchild->d_inode->i_ino;
159 child_res_id.name[1] = dchild->d_inode->i_generation;
162 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
163 child_res_id, LDLM_PLAIN, NULL, 0, child_mode,
164 &lock_flags, ldlm_completion_ast,
165 mds_blocking_ast, NULL, NULL, child_lockh);
166 if (rc != ELDLM_OK) {
167 CERROR("ldlm_cli_enqueue: %d\n", rc);
168 GOTO(out_step_3, rc = -EIO);
171 mds_pack_inode2fid(&body->fid1, dchild->d_inode);
172 mds_pack_inode2body(body, dchild->d_inode);
173 if (S_ISREG(dchild->d_inode->i_mode)) {
174 rc = mds_pack_md(obd, req->rq_repmsg, 2, body, dchild->d_inode);
176 GOTO(out_step_4, rc);
178 /* If this isn't a regular file, we can't open it. */
179 GOTO(out_step_3, rc = 0); /* returns the lock to the client */
182 if (!created && (rec->ur_flags & O_CREAT) && (rec->ur_flags & O_EXCL)) {
183 /* File already exists, we didn't just create it, and we
184 * were passed O_EXCL; err-or. */
185 GOTO(out_step_3, rc = -EEXIST); // returns a lock to the client
188 /* If we're opening a file without an EA, the client needs a write
190 if (child_mode != LCK_PW && S_ISREG(dchild->d_inode->i_mode) &&
191 !(body->valid & OBD_MD_FLEASIZE)) {
192 ldlm_lock_decref(child_lockh, child_mode);
197 /* Step 5: Open it */
198 rep->lock_policy_res1 |= IT_OPEN_OPEN;
199 mfd = kmem_cache_alloc(mds_file_cache, GFP_KERNEL);
201 CERROR("mds: out of memory\n");
202 GOTO(out_step_4, req->rq_status = -ENOMEM);
205 /* dentry_open does a dput(de) and mntput(mds->mds_vfsmnt) on error */
206 mntget(mds->mds_vfsmnt);
207 file = dentry_open(dchild,mds->mds_vfsmnt,
208 rec->ur_flags & ~(O_DIRECT | O_TRUNC));
210 GOTO(out_step_5, rc = PTR_ERR(file));
212 file->private_data = mfd;
213 mfd->mfd_file = file;
214 get_random_bytes(&mfd->mfd_servercookie, sizeof(mfd->mfd_servercookie));
215 spin_lock(&med->med_open_lock);
216 list_add(&mfd->mfd_list, &med->med_open_head);
217 spin_unlock(&med->med_open_lock);
219 body->handle.addr = (__u64)(unsigned long)mfd;
220 body->handle.cookie = mfd->mfd_servercookie;
221 CDEBUG(D_INODE, "file %p: mfd %p, cookie "LPX64"\n",
222 mfd->mfd_file, mfd, mfd->mfd_servercookie);
223 GOTO(out_step_2, rc = 0); /* returns a lock to the client */
227 kmem_cache_free(mds_file_cache, mfd);
231 ldlm_lock_decref(child_lockh, child_mode);
236 ldlm_lock_decref(&parent_lockh, parent_mode);