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_rcvd,
60 __u32 *last_xid, 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_rcvd = req->rq_repmsg->last_rcvd;
85 *last_xid = body->last_xid;
87 CDEBUG(D_NET, "root ino=%ld, last_committed=%Lu, last_rcvd=%Lu,"
89 (unsigned long)rootfid->id,
90 (unsigned long long)*last_committed,
91 (unsigned long long)*last_rcvd,
101 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
102 uuid_t **uuids, struct ptlrpc_request **request)
104 struct ptlrpc_request *req;
105 struct mds_status_req *streq;
106 struct lov_obd *lov = &obd->u.lov;
107 struct client_obd *mdc = &lov->mdcobd->u.cli;
108 struct lov_desc *desc = &lov->desc;
109 int rc, size[2] = {sizeof(*streq)};
112 req = ptlrpc_prep_req2(mdc_connh, MDS_GETLOVINFO, 1, size, NULL);
114 GOTO(out, rc = -ENOMEM);
117 streq = lustre_msg_buf(req->rq_reqmsg, 0);
118 streq->flags = HTON__u32(MDS_STATUS_LOV);
119 streq->repbuf = HTON__u32(8000);
121 /* prepare for reply */
122 req->rq_level = LUSTRE_CONN_CON;
123 size[0] = sizeof(*desc);
125 req->rq_replen = lustre_msg_size(2, size);
127 rc = ptlrpc_queue_wait(req);
128 rc = ptlrpc_check_status(req, rc);
131 memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
132 *uuids = lustre_msg_buf(req->rq_repmsg, 1);
133 lov_unpackdesc(desc);
135 mdc->cl_max_mdsize = sizeof(*desc) +
136 desc->ld_tgt_count * sizeof(uuid_t);
144 int mdc_getattr(struct lustre_handle *conn,
145 obd_id ino, int type, unsigned long valid, size_t ea_size,
146 struct ptlrpc_request **request)
148 struct ptlrpc_client *cl;
149 struct ptlrpc_connection *connection;
150 struct lustre_handle *rconn;
151 struct ptlrpc_request *req;
152 struct mds_body *body;
153 int rc, size[2] = {sizeof(*body), 0}, bufcount = 1;
156 mdc_con2cl(conn, &cl, &connection, &rconn);
157 req = ptlrpc_prep_req2(conn, MDS_GETATTR, 1, size, NULL);
159 GOTO(out, rc = -ENOMEM);
161 body = lustre_msg_buf(req->rq_reqmsg, 0);
162 ll_ino2fid(&body->fid1, ino, 0, type);
166 struct client_obd *mdc = &class_conn2obd(conn)->u.cli;
168 size[1] = mdc->cl_max_mdsize;
169 } else if (valid & OBD_MD_LINKNAME) {
173 req->rq_replen = lustre_msg_size(bufcount, size);
175 rc = ptlrpc_queue_wait(req);
176 rc = ptlrpc_check_status(req, rc);
179 body = lustre_msg_buf(req->rq_repmsg, 0);
180 mds_unpack_body(body);
181 CDEBUG(D_NET, "mode: %o\n", body->mode);
190 static int mdc_lock_callback(struct lustre_handle *lockh,
191 struct ldlm_lock_desc *desc, void *data,
192 int data_len, struct ptlrpc_request **req)
195 struct inode *inode = data;
199 /* Completion AST. Do nothing. */
203 if (data_len != sizeof(*inode)) {
204 CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
210 /* FIXME: do something better than throwing away everything */
213 if (S_ISDIR(inode->i_mode)) {
214 CDEBUG(D_INODE, "invalidating inode %ld\n", inode->i_ino);
215 invalidate_inode_pages(inode);
218 rc = ldlm_cli_cancel(lockh);
220 CERROR("ldlm_cli_cancel: %d\n", rc);
226 int mdc_enqueue(struct lustre_handle *conn, int lock_type,
227 struct lookup_intent *it, int lock_mode, struct inode *dir,
228 struct dentry *de, struct lustre_handle *lockh, __u64 id,
229 char *tgt, int tgtlen, void *data, int datalen)
231 struct ptlrpc_request *req;
232 struct obd_device *obddev = class_conn2obd(conn);
233 __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
234 int size[5] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
236 int repsize[3] = {sizeof(struct ldlm_reply),
237 sizeof(struct mds_body),
238 obddev->u.cli.cl_max_mdsize};
239 struct ldlm_reply *dlm_rep;
240 struct ldlm_intent *lit;
243 LDLM_DEBUG_NOLOCK("mdsintent %s dir %ld", ldlm_it2str(it->it_op), dir->i_ino);
247 it->it_mode = (it->it_mode | S_IFDIR) & ~current->fs->umask;
249 case (IT_CREAT|IT_OPEN):
251 it->it_mode |= S_IFREG; /* no break */
253 it->it_mode &= ~current->fs->umask;
256 it->it_mode = (it->it_mode | S_IFLNK) & ~current->fs->umask;
260 if (it->it_op & (IT_MKDIR | IT_CREAT | IT_SYMLINK | IT_MKNOD)) {
261 size[2] = sizeof(struct mds_rec_create);
262 size[3] = de->d_name.len + 1;
263 size[4] = tgtlen + 1;
264 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
268 /* pack the intent */
269 lit = lustre_msg_buf(req->rq_reqmsg, 1);
270 lit->opc = NTOH__u64((__u64)it->it_op);
272 /* pack the intended request */
273 mds_create_pack(req, 2, dir, it->it_mode, id, current->fsuid,
274 current->fsgid, CURRENT_TIME, de->d_name.name,
275 de->d_name.len, tgt, tgtlen);
276 req->rq_replen = lustre_msg_size(3, repsize);
277 } else if (it->it_op == IT_RENAME2) {
278 struct dentry *old_de = it->it_data;
280 size[2] = sizeof(struct mds_rec_rename);
281 size[3] = old_de->d_name.len + 1;
282 size[4] = de->d_name.len + 1;
283 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
287 /* pack the intent */
288 lit = lustre_msg_buf(req->rq_reqmsg, 1);
289 lit->opc = NTOH__u64((__u64)it->it_op);
291 /* pack the intended request */
292 mds_rename_pack(req, 2, old_de->d_parent->d_inode, dir,
293 old_de->d_name.name, old_de->d_name.len,
294 de->d_name.name, de->d_name.len);
295 req->rq_replen = lustre_msg_size(1, repsize);
296 } else if (it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) {
297 size[2] = sizeof(struct mds_rec_unlink);
298 size[3] = de->d_name.len + 1;
299 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
303 /* pack the intent */
304 lit = lustre_msg_buf(req->rq_reqmsg, 1);
305 lit->opc = NTOH__u64((__u64)it->it_op);
307 /* pack the intended request */
308 mds_unlink_pack(req, 2, dir, NULL, de->d_name.name,
311 req->rq_replen = lustre_msg_size(3, repsize);
312 } else if (it->it_op == IT_GETATTR || it->it_op == IT_RENAME ||
313 it->it_op == IT_OPEN || it->it_op == IT_SETATTR ||
314 it->it_op == IT_LOOKUP || it->it_op == IT_READLINK) {
315 size[2] = sizeof(struct mds_body);
316 size[3] = de->d_name.len + 1;
318 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
322 /* pack the intent */
323 lit = lustre_msg_buf(req->rq_reqmsg, 1);
324 lit->opc = NTOH__u64((__u64)it->it_op);
326 /* pack the intended request */
327 mds_getattr_pack(req, 2, dir, de->d_name.name, de->d_name.len);
329 /* get ready for the reply */
330 req->rq_replen = lustre_msg_size(3, repsize);
331 } else if (it->it_op == IT_READDIR) {
332 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 1, size, NULL);
336 /* get ready for the reply */
337 req->rq_replen = lustre_msg_size(1, repsize);
342 #warning FIXME: the data here needs to be different if a lock was granted for a different inode
343 rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id, lock_type,
344 NULL, 0, lock_mode, &flags, ldlm_completion_ast,
345 (void *)mdc_lock_callback, data, datalen, lockh);
346 if (rc == -ENOENT || rc == ELDLM_LOCK_ABORTED) {
348 memset(lockh, 0, sizeof(*lockh));
350 } else if (rc != 0) {
351 CERROR("ldlm_cli_enqueue: %d\n", rc);
355 dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
356 it->it_disposition = (int) dlm_rep->lock_policy_res1;
357 it->it_status = (int) dlm_rep->lock_policy_res2;
358 it->it_lock_mode = lock_mode;
364 int mdc_open(struct lustre_handle *conn, obd_id ino, int type, int flags,
365 struct lov_stripe_md *smd, __u64 cookie, __u64 *fh,
366 struct ptlrpc_request **request)
368 struct ptlrpc_client *cl;
369 struct ptlrpc_connection *connection;
370 struct lustre_handle *rconn;
371 struct mds_body *body;
372 int rc, size[2] = {sizeof(*body)}, bufcount = 1;
373 struct ptlrpc_request *req;
378 size[1] = smd->lmd_easize;
381 mdc_con2cl(conn, &cl, &connection, &rconn);
382 req = ptlrpc_prep_req2(conn, MDS_OPEN, bufcount, size, NULL);
384 GOTO(out, rc = -ENOMEM);
386 req->rq_flags |= PTL_RPC_FL_REPLAY;
387 body = lustre_msg_buf(req->rq_reqmsg, 0);
389 ll_ino2fid(&body->fid1, ino, 0, type);
390 body->flags = HTON__u32(flags);
391 body->extra = cookie;
394 memcpy(lustre_msg_buf(req->rq_reqmsg, 1), smd, smd->lmd_easize);
396 req->rq_replen = lustre_msg_size(1, size);
398 rc = ptlrpc_queue_wait(req);
399 rc = ptlrpc_check_status(req, rc);
401 body = lustre_msg_buf(req->rq_repmsg, 0);
402 mds_unpack_body(body);
412 int mdc_close(struct lustre_handle *conn,
413 obd_id ino, int type, __u64 fh, struct ptlrpc_request **request)
415 struct mds_body *body;
416 int rc, size = sizeof(*body);
417 struct ptlrpc_request *req;
419 req = ptlrpc_prep_req2(conn, MDS_CLOSE, 1, &size, NULL);
421 GOTO(out, rc = -ENOMEM);
423 body = lustre_msg_buf(req->rq_reqmsg, 0);
424 ll_ino2fid(&body->fid1, ino, 0, type);
427 req->rq_replen = lustre_msg_size(0, NULL);
429 rc = ptlrpc_queue_wait(req);
430 rc = ptlrpc_check_status(req, rc);
438 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
439 char *addr, struct ptlrpc_request **request)
441 struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
442 struct ptlrpc_request *req = NULL;
443 struct ptlrpc_bulk_desc *desc = NULL;
444 struct ptlrpc_bulk_page *bulk = NULL;
445 struct mds_body *body;
446 int rc, size = sizeof(*body);
449 CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
451 desc = ptlrpc_prep_bulk(connection);
453 GOTO(out, rc = -ENOMEM);
455 req = ptlrpc_prep_req2(conn, MDS_READPAGE, 1, &size, NULL);
457 GOTO(out2, rc = -ENOMEM);
459 bulk = ptlrpc_prep_bulk_page(desc);
460 bulk->b_buflen = PAGE_SIZE;
462 bulk->b_xid = req->rq_xid;
463 desc->b_portal = MDS_BULK_PORTAL;
465 rc = ptlrpc_register_bulk(desc);
467 CERROR("couldn't setup bulk sink: error %d.\n", rc);
471 body = lustre_msg_buf(req->rq_reqmsg, 0);
473 body->fid1.f_type = type;
476 req->rq_replen = lustre_msg_size(1, &size);
477 rc = ptlrpc_queue_wait(req);
478 rc = ptlrpc_check_status(req, rc);
480 ptlrpc_abort_bulk(desc);
483 body = lustre_msg_buf(req->rq_repmsg, 0);
484 mds_unpack_body(body);
489 ptlrpc_free_bulk(desc);
495 int mdc_statfs(struct lustre_handle *conn, struct statfs *sfs,
496 struct ptlrpc_request **request)
498 struct obd_statfs *osfs;
499 struct ptlrpc_request *req;
500 int rc, size = sizeof(*osfs);
503 req = ptlrpc_prep_req2(conn, MDS_STATFS, 0, NULL, NULL);
505 GOTO(out, rc = -ENOMEM);
506 req->rq_replen = lustre_msg_size(1, &size);
508 rc = ptlrpc_queue_wait(req);
509 rc = ptlrpc_check_status(req, rc);
514 osfs = lustre_msg_buf(req->rq_repmsg, 0);
515 obd_statfs_unpack(osfs, sfs);
524 struct obd_ops mdc_obd_ops = {
525 o_setup: client_obd_setup,
526 o_cleanup: client_obd_cleanup,
527 o_connect: client_obd_connect,
528 o_disconnect: client_obd_disconnect,
531 static int __init ptlrpc_request_init(void)
533 return class_register_type(&mdc_obd_ops, LUSTRE_MDC_NAME);
536 static void __exit ptlrpc_request_exit(void)
538 class_unregister_type(LUSTRE_MDC_NAME);
541 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
542 MODULE_DESCRIPTION("Lustre Metadata Client v1.0");
543 MODULE_LICENSE("GPL");
545 EXPORT_SYMBOL(mdc_getstatus);
546 EXPORT_SYMBOL(mdc_getlovinfo);
547 EXPORT_SYMBOL(mdc_enqueue);
548 EXPORT_SYMBOL(mdc_getattr);
549 EXPORT_SYMBOL(mdc_statfs);
550 EXPORT_SYMBOL(mdc_create);
551 EXPORT_SYMBOL(mdc_unlink);
552 EXPORT_SYMBOL(mdc_rename);
553 EXPORT_SYMBOL(mdc_link);
554 EXPORT_SYMBOL(mdc_readpage);
555 EXPORT_SYMBOL(mdc_setattr);
556 EXPORT_SYMBOL(mdc_close);
557 EXPORT_SYMBOL(mdc_open);
559 module_init(ptlrpc_request_init);
560 module_exit(ptlrpc_request_exit);