1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
6 * This file is part of Lustre, http://www.sf.net/projects/lustre/
8 * Lustre is free software; you can redistribute it and/or
9 * modify it under the terms of version 2 of the GNU General Public
10 * License as published by the Free Software Foundation.
12 * Lustre is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
17 * You should have received a copy of the GNU General Public License
18 * along with Lustre; if not, write to the Free Software
19 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24 #define DEBUG_SUBSYSTEM S_MDC
26 #include <linux/module.h>
27 #include <linux/miscdevice.h>
28 #include <linux/lustre_mds.h>
29 #include <linux/lustre_lite.h>
30 #include <linux/lustre_dlm.h>
31 #include <linux/init.h>
32 #include <linux/obd_lov.h>
34 #define REQUEST_MINOR 244
36 extern int mds_queue_req(struct ptlrpc_request *);
38 int mdc_con2cl(struct lustre_handle *conn, struct ptlrpc_client **cl,
39 struct ptlrpc_connection **connection,
40 struct lustre_handle **rconn)
42 struct obd_export *export;
43 struct client_obd *mdc;
45 export = class_conn2export(conn);
49 mdc = &export->exp_obd->u.cli;
52 *connection = mdc->cl_conn;
53 *rconn = &export->exp_rconnh;
58 int mdc_getstatus(struct lustre_handle *conn, struct ll_fid *rootfid,
59 __u64 *last_committed, __u64 *last_xid,
60 struct ptlrpc_request **request)
62 struct ptlrpc_request *req;
63 struct mds_body *body;
64 int rc, size = sizeof(*body);
67 req = ptlrpc_prep_req2(conn, MDS_GETSTATUS, 1, &size, NULL);
69 GOTO(out, rc = -ENOMEM);
71 body = lustre_msg_buf(req->rq_reqmsg, 0);
72 req->rq_level = LUSTRE_CONN_CON;
73 req->rq_replen = lustre_msg_size(1, &size);
75 mds_pack_req_body(req);
76 rc = ptlrpc_queue_wait(req);
77 rc = ptlrpc_check_status(req, rc);
80 body = lustre_msg_buf(req->rq_repmsg, 0);
81 mds_unpack_body(body);
82 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
83 *last_committed = req->rq_repmsg->last_committed;
84 *last_xid = req->rq_repmsg->last_xid;
86 CDEBUG(D_NET,"root ino=%ld, last_committed=%Lu, last_xid=%Ld\n",
87 (unsigned long)rootfid->id,
88 (unsigned long long)*last_committed,
89 (unsigned long long)*last_xid);
98 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
99 uuid_t **uuids, struct ptlrpc_request **request)
101 struct ptlrpc_request *req;
102 struct mds_status_req *streq;
103 struct lov_obd *lov = &obd->u.lov;
104 struct client_obd *mdc = &lov->mdcobd->u.cli;
105 struct lov_desc *desc = &lov->desc;
106 int rc, size[2] = {sizeof(*streq)};
109 req = ptlrpc_prep_req2(mdc_connh, MDS_GETLOVINFO, 1, size, NULL);
111 GOTO(out, rc = -ENOMEM);
114 streq = lustre_msg_buf(req->rq_reqmsg, 0);
115 streq->flags = HTON__u32(MDS_STATUS_LOV);
116 streq->repbuf = HTON__u32(8000);
118 /* prepare for reply */
119 req->rq_level = LUSTRE_CONN_CON;
120 size[0] = sizeof(*desc);
122 req->rq_replen = lustre_msg_size(2, size);
124 rc = ptlrpc_queue_wait(req);
125 rc = ptlrpc_check_status(req, rc);
128 memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
129 *uuids = lustre_msg_buf(req->rq_repmsg, 1);
130 lov_unpackdesc(desc);
131 mdc->cl_max_mdsize = sizeof(struct lov_mds_md) +
132 desc->ld_tgt_count * sizeof(struct lov_object_id);
141 int mdc_getattr(struct lustre_handle *conn,
142 obd_id ino, int type, unsigned long valid, size_t ea_size,
143 struct ptlrpc_request **request)
145 struct ptlrpc_request *req;
146 struct mds_body *body;
147 int rc, size[2] = {sizeof(*body), 0}, bufcount = 1;
150 req = ptlrpc_prep_req2(conn, MDS_GETATTR, 1, size, NULL);
152 GOTO(out, rc = -ENOMEM);
154 body = lustre_msg_buf(req->rq_reqmsg, 0);
155 ll_ino2fid(&body->fid1, ino, 0, type);
159 struct client_obd *mdc = &class_conn2obd(conn)->u.cli;
161 size[1] = mdc->cl_max_mdsize;
162 } else if (valid & OBD_MD_LINKNAME) {
166 req->rq_replen = lustre_msg_size(bufcount, size);
168 rc = ptlrpc_queue_wait(req);
169 rc = ptlrpc_check_status(req, rc);
172 body = lustre_msg_buf(req->rq_repmsg, 0);
173 mds_unpack_body(body);
174 CDEBUG(D_NET, "mode: %o\n", body->mode);
183 static int mdc_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
184 void *data, __u32 data_len)
187 struct inode *inode = data;
188 struct lustre_handle lockh;
191 if (data_len != sizeof(*inode)) {
192 CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
198 /* FIXME: do something better than throwing away everything */
201 if (S_ISDIR(inode->i_mode)) {
202 CDEBUG(D_INODE, "invalidating inode %ld\n", inode->i_ino);
203 invalidate_inode_pages(inode);
206 ldlm_lock2handle(lock, &lockh);
207 rc = ldlm_cli_cancel(&lockh);
209 CERROR("ldlm_cli_cancel: %d\n", rc);
215 int mdc_enqueue(struct lustre_handle *conn, int lock_type,
216 struct lookup_intent *it, int lock_mode, struct inode *dir,
217 struct dentry *de, struct lustre_handle *lockh,
218 char *tgt, int tgtlen, void *data, int datalen)
220 struct ptlrpc_request *req;
221 struct obd_device *obddev = class_conn2obd(conn);
222 __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
223 int size[5] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
225 int repsize[3] = {sizeof(struct ldlm_reply),
226 sizeof(struct mds_body),
227 obddev->u.cli.cl_max_mdsize};
228 struct ldlm_reply *dlm_rep;
229 struct ldlm_intent *lit;
232 LDLM_DEBUG_NOLOCK("mdsintent %s dir %ld", ldlm_it2str(it->it_op),
237 it->it_mode = (it->it_mode | S_IFDIR) & ~current->fs->umask;
239 case (IT_CREAT|IT_OPEN):
241 it->it_mode |= S_IFREG; /* no break */
243 it->it_mode &= ~current->fs->umask;
246 it->it_mode = (it->it_mode | S_IFLNK) & ~current->fs->umask;
250 if (it->it_op & (IT_MKDIR | IT_CREAT | IT_SYMLINK | IT_MKNOD)) {
251 size[2] = sizeof(struct mds_rec_create);
252 size[3] = de->d_name.len + 1;
253 size[4] = tgtlen + 1;
254 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
258 /* pack the intent */
259 lit = lustre_msg_buf(req->rq_reqmsg, 1);
260 lit->opc = NTOH__u64((__u64)it->it_op);
262 /* pack the intended request */
263 mds_create_pack(req, 2, dir, it->it_mode, 0, current->fsuid,
264 current->fsgid, CURRENT_TIME, de->d_name.name,
265 de->d_name.len, tgt, tgtlen);
266 req->rq_replen = lustre_msg_size(3, repsize);
267 } else if (it->it_op == IT_RENAME2) {
268 struct dentry *old_de = it->it_data;
270 size[2] = sizeof(struct mds_rec_rename);
271 size[3] = old_de->d_name.len + 1;
272 size[4] = de->d_name.len + 1;
273 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
277 /* pack the intent */
278 lit = lustre_msg_buf(req->rq_reqmsg, 1);
279 lit->opc = NTOH__u64((__u64)it->it_op);
281 /* pack the intended request */
282 mds_rename_pack(req, 2, old_de->d_parent->d_inode, dir,
283 old_de->d_name.name, old_de->d_name.len,
284 de->d_name.name, de->d_name.len);
285 req->rq_replen = lustre_msg_size(3, repsize);
286 } else if (it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) {
287 size[2] = sizeof(struct mds_rec_unlink);
288 size[3] = de->d_name.len + 1;
289 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
293 /* pack the intent */
294 lit = lustre_msg_buf(req->rq_reqmsg, 1);
295 lit->opc = NTOH__u64((__u64)it->it_op);
297 /* pack the intended request */
298 mds_unlink_pack(req, 2, dir, NULL,
299 it->it_op == IT_UNLINK ? S_IFREG : S_IFDIR,
300 de->d_name.name, de->d_name.len);
302 req->rq_replen = lustre_msg_size(3, repsize);
303 } else if (it->it_op == IT_GETATTR || it->it_op == IT_RENAME ||
304 it->it_op == IT_OPEN || it->it_op == IT_SETATTR ||
305 it->it_op == IT_LOOKUP || it->it_op == IT_READLINK) {
306 size[2] = sizeof(struct mds_body);
307 size[3] = de->d_name.len + 1;
309 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
313 /* pack the intent */
314 lit = lustre_msg_buf(req->rq_reqmsg, 1);
315 lit->opc = NTOH__u64((__u64)it->it_op);
317 /* pack the intended request */
318 mds_getattr_pack(req, 2, dir, de->d_name.name, de->d_name.len);
320 /* get ready for the reply */
321 req->rq_replen = lustre_msg_size(3, repsize);
322 } else if (it->it_op == IT_READDIR) {
323 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 1, size, NULL);
327 /* get ready for the reply */
328 req->rq_replen = lustre_msg_size(1, repsize);
333 #warning FIXME: the data here needs to be different if a lock was granted for a different inode
334 rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id,
335 lock_type, NULL, 0, lock_mode, &flags,
336 ldlm_completion_ast, mdc_blocking_ast, data,
339 /* This can go when we're sure that this can never happen */
342 if (rc == ELDLM_LOCK_ABORTED) {
344 memset(lockh, 0, sizeof(*lockh));
346 } else if (rc != 0) {
347 CERROR("ldlm_cli_enqueue: %d\n", rc);
351 dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
352 it->it_disposition = (int) dlm_rep->lock_policy_res1;
353 it->it_status = (int) dlm_rep->lock_policy_res2;
354 it->it_lock_mode = lock_mode;
360 int mdc_open(struct lustre_handle *conn, obd_id ino, int type, int flags,
361 struct lov_stripe_md *smd, __u64 cookie, __u64 *fh,
362 struct ptlrpc_request **request)
364 struct ptlrpc_client *cl;
365 struct ptlrpc_connection *connection;
366 struct lustre_handle *rconn;
367 struct mds_body *body;
368 int rc, size[2] = {sizeof(*body)}, bufcount = 1;
369 struct ptlrpc_request *req;
374 size[1] = smd->lmd_easize;
377 mdc_con2cl(conn, &cl, &connection, &rconn);
378 req = ptlrpc_prep_req2(conn, MDS_OPEN, bufcount, size, NULL);
380 GOTO(out, rc = -ENOMEM);
382 req->rq_flags |= PTL_RPC_FL_REPLAY;
383 body = lustre_msg_buf(req->rq_reqmsg, 0);
385 ll_ino2fid(&body->fid1, ino, 0, type);
386 body->flags = HTON__u32(flags);
387 body->extra = cookie;
390 lov_packmd(lustre_msg_buf(req->rq_reqmsg, 1), smd);
392 req->rq_replen = lustre_msg_size(1, size);
394 rc = ptlrpc_queue_wait(req);
395 rc = ptlrpc_check_status(req, rc);
397 body = lustre_msg_buf(req->rq_repmsg, 0);
398 mds_unpack_body(body);
408 int mdc_close(struct lustre_handle *conn,
409 obd_id ino, int type, __u64 fh, struct ptlrpc_request **request)
411 struct mds_body *body;
412 int rc, size = sizeof(*body);
413 struct ptlrpc_request *req;
415 req = ptlrpc_prep_req2(conn, MDS_CLOSE, 1, &size, NULL);
417 GOTO(out, rc = -ENOMEM);
419 body = lustre_msg_buf(req->rq_reqmsg, 0);
420 ll_ino2fid(&body->fid1, ino, 0, type);
423 req->rq_replen = lustre_msg_size(0, NULL);
425 rc = ptlrpc_queue_wait(req);
426 rc = ptlrpc_check_status(req, rc);
434 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
435 char *addr, struct ptlrpc_request **request)
437 struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
438 struct ptlrpc_request *req = NULL;
439 struct ptlrpc_bulk_desc *desc = NULL;
440 struct ptlrpc_bulk_page *bulk = NULL;
441 struct mds_body *body;
442 int rc, size = sizeof(*body);
445 CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
447 desc = ptlrpc_prep_bulk(connection);
449 GOTO(out, rc = -ENOMEM);
451 req = ptlrpc_prep_req2(conn, MDS_READPAGE, 1, &size, NULL);
453 GOTO(out2, rc = -ENOMEM);
455 bulk = ptlrpc_prep_bulk_page(desc);
456 bulk->b_buflen = PAGE_SIZE;
458 bulk->b_xid = req->rq_xid;
459 desc->b_portal = MDS_BULK_PORTAL;
461 rc = ptlrpc_register_bulk(desc);
463 CERROR("couldn't setup bulk sink: error %d.\n", rc);
467 body = lustre_msg_buf(req->rq_reqmsg, 0);
469 body->fid1.f_type = type;
472 req->rq_replen = lustre_msg_size(1, &size);
473 rc = ptlrpc_queue_wait(req);
474 rc = ptlrpc_check_status(req, rc);
476 ptlrpc_abort_bulk(desc);
479 body = lustre_msg_buf(req->rq_repmsg, 0);
480 mds_unpack_body(body);
485 ptlrpc_free_bulk(desc);
491 int mdc_statfs(struct lustre_handle *conn, struct statfs *sfs,
492 struct ptlrpc_request **request)
494 struct obd_statfs *osfs;
495 struct ptlrpc_request *req;
496 int rc, size = sizeof(*osfs);
499 req = ptlrpc_prep_req2(conn, MDS_STATFS, 0, NULL, NULL);
501 GOTO(out, rc = -ENOMEM);
502 req->rq_replen = lustre_msg_size(1, &size);
504 rc = ptlrpc_queue_wait(req);
505 rc = ptlrpc_check_status(req, rc);
510 osfs = lustre_msg_buf(req->rq_repmsg, 0);
511 obd_statfs_unpack(osfs, sfs);
520 struct obd_ops mdc_obd_ops = {
521 o_setup: client_obd_setup,
522 o_cleanup: client_obd_cleanup,
523 o_connect: client_obd_connect,
524 o_disconnect: client_obd_disconnect,
527 static int __init ptlrpc_request_init(void)
529 return class_register_type(&mdc_obd_ops, LUSTRE_MDC_NAME);
532 static void __exit ptlrpc_request_exit(void)
534 class_unregister_type(LUSTRE_MDC_NAME);
537 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
538 MODULE_DESCRIPTION("Lustre Metadata Client v1.0");
539 MODULE_LICENSE("GPL");
541 EXPORT_SYMBOL(mdc_getstatus);
542 EXPORT_SYMBOL(mdc_getlovinfo);
543 EXPORT_SYMBOL(mdc_enqueue);
544 EXPORT_SYMBOL(mdc_getattr);
545 EXPORT_SYMBOL(mdc_statfs);
546 EXPORT_SYMBOL(mdc_create);
547 EXPORT_SYMBOL(mdc_unlink);
548 EXPORT_SYMBOL(mdc_rename);
549 EXPORT_SYMBOL(mdc_link);
550 EXPORT_SYMBOL(mdc_readpage);
551 EXPORT_SYMBOL(mdc_setattr);
552 EXPORT_SYMBOL(mdc_close);
553 EXPORT_SYMBOL(mdc_open);
555 module_init(ptlrpc_request_init);
556 module_exit(ptlrpc_request_exit);