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_getstatus(struct lustre_handle *conn, struct ll_fid *rootfid,
39 __u64 *last_committed, __u64 *last_xid,
40 struct ptlrpc_request **request)
42 struct ptlrpc_request *req;
43 struct mds_body *body;
44 int rc, size = sizeof(*body);
47 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_GETSTATUS, 1, &size,
50 GOTO(out, rc = -ENOMEM);
52 body = lustre_msg_buf(req->rq_reqmsg, 0);
53 req->rq_level = LUSTRE_CONN_CON;
54 req->rq_replen = lustre_msg_size(1, &size);
56 mds_pack_req_body(req);
57 rc = ptlrpc_queue_wait(req);
58 rc = ptlrpc_check_status(req, rc);
61 body = lustre_msg_buf(req->rq_repmsg, 0);
62 mds_unpack_body(body);
63 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
64 *last_committed = req->rq_repmsg->last_committed;
65 *last_xid = req->rq_repmsg->last_xid;
67 CDEBUG(D_NET,"root ino=%ld, last_committed=%Lu, last_xid=%Ld\n",
68 (unsigned long)rootfid->id,
69 (unsigned long long)*last_committed,
70 (unsigned long long)*last_xid);
79 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
80 uuid_t **uuids, struct ptlrpc_request **request)
82 struct ptlrpc_request *req;
83 struct mds_status_req *streq;
84 struct lov_obd *lov = &obd->u.lov;
85 struct client_obd *mdc = &lov->mdcobd->u.cli;
86 struct lov_desc *desc = &lov->desc;
87 int rc, size[2] = {sizeof(*streq)};
90 req = ptlrpc_prep_req(class_conn2cliimp(mdc_connh), MDS_GETLOVINFO, 1,
93 GOTO(out, rc = -ENOMEM);
96 streq = lustre_msg_buf(req->rq_reqmsg, 0);
97 streq->flags = HTON__u32(MDS_STATUS_LOV);
98 streq->repbuf = HTON__u32(8000);
100 /* prepare for reply */
101 req->rq_level = LUSTRE_CONN_CON;
102 size[0] = sizeof(*desc);
104 req->rq_replen = lustre_msg_size(2, size);
106 rc = ptlrpc_queue_wait(req);
107 rc = ptlrpc_check_status(req, rc);
110 memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
111 *uuids = lustre_msg_buf(req->rq_repmsg, 1);
112 lov_unpackdesc(desc);
113 mdc->cl_max_mdsize = sizeof(struct lov_mds_md) +
114 desc->ld_tgt_count * sizeof(struct lov_object_id);
123 int mdc_getattr(struct lustre_handle *conn,
124 obd_id ino, int type, unsigned long valid, size_t ea_size,
125 struct ptlrpc_request **request)
127 struct ptlrpc_request *req;
128 struct mds_body *body;
129 int rc, size[2] = {sizeof(*body), 0}, bufcount = 1;
132 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_GETATTR, 1, size,
135 GOTO(out, rc = -ENOMEM);
137 body = lustre_msg_buf(req->rq_reqmsg, 0);
138 ll_ino2fid(&body->fid1, ino, 0, type);
142 struct client_obd *mdc = &class_conn2obd(conn)->u.cli;
144 size[1] = mdc->cl_max_mdsize;
145 } else if (valid & OBD_MD_LINKNAME) {
149 req->rq_replen = lustre_msg_size(bufcount, size);
151 rc = ptlrpc_queue_wait(req);
152 rc = ptlrpc_check_status(req, rc);
155 body = lustre_msg_buf(req->rq_repmsg, 0);
156 mds_unpack_body(body);
157 CDEBUG(D_NET, "mode: %o\n", body->mode);
166 static int mdc_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
167 void *data, __u32 data_len)
170 struct inode *inode = data;
171 struct lustre_handle lockh;
174 if (data_len != sizeof(*inode)) {
175 CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
181 /* FIXME: do something better than throwing away everything */
184 if (S_ISDIR(inode->i_mode)) {
185 CDEBUG(D_INODE, "invalidating inode %ld\n", inode->i_ino);
186 invalidate_inode_pages(inode);
189 ldlm_lock2handle(lock, &lockh);
190 rc = ldlm_cli_cancel(&lockh);
192 CERROR("ldlm_cli_cancel: %d\n", rc);
198 int mdc_enqueue(struct lustre_handle *conn, int lock_type,
199 struct lookup_intent *it, int lock_mode, struct inode *dir,
200 struct dentry *de, struct lustre_handle *lockh,
201 char *tgt, int tgtlen, void *data, int datalen)
203 struct ptlrpc_request *req;
204 struct obd_device *obddev = class_conn2obd(conn);
205 __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
206 int size[5] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
208 int repsize[3] = {sizeof(struct ldlm_reply),
209 sizeof(struct mds_body),
210 obddev->u.cli.cl_max_mdsize};
211 struct ldlm_reply *dlm_rep;
212 struct ldlm_intent *lit;
215 LDLM_DEBUG_NOLOCK("mdsintent %s dir %ld", ldlm_it2str(it->it_op),
220 it->it_mode = (it->it_mode | S_IFDIR) & ~current->fs->umask;
222 case (IT_CREAT|IT_OPEN):
224 it->it_mode |= S_IFREG; /* no break */
226 it->it_mode &= ~current->fs->umask;
229 it->it_mode = (it->it_mode | S_IFLNK) & ~current->fs->umask;
233 if (it->it_op & (IT_MKDIR | IT_CREAT | IT_SYMLINK | IT_MKNOD)) {
234 size[2] = sizeof(struct mds_rec_create);
235 size[3] = de->d_name.len + 1;
236 size[4] = tgtlen + 1;
237 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 5,
242 /* pack the intent */
243 lit = lustre_msg_buf(req->rq_reqmsg, 1);
244 lit->opc = NTOH__u64((__u64)it->it_op);
246 /* pack the intended request */
247 mds_create_pack(req, 2, dir, it->it_mode, 0, current->fsuid,
248 current->fsgid, CURRENT_TIME, de->d_name.name,
249 de->d_name.len, tgt, tgtlen);
250 req->rq_replen = lustre_msg_size(3, repsize);
251 } else if (it->it_op == IT_RENAME2) {
252 struct dentry *old_de = it->it_data;
254 size[2] = sizeof(struct mds_rec_rename);
255 size[3] = old_de->d_name.len + 1;
256 size[4] = de->d_name.len + 1;
257 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 5,
262 /* pack the intent */
263 lit = lustre_msg_buf(req->rq_reqmsg, 1);
264 lit->opc = NTOH__u64((__u64)it->it_op);
266 /* pack the intended request */
267 mds_rename_pack(req, 2, old_de->d_parent->d_inode, dir,
268 old_de->d_name.name, old_de->d_name.len,
269 de->d_name.name, de->d_name.len);
270 req->rq_replen = lustre_msg_size(3, repsize);
271 } else if (it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) {
272 size[2] = sizeof(struct mds_rec_unlink);
273 size[3] = de->d_name.len + 1;
274 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
279 /* pack the intent */
280 lit = lustre_msg_buf(req->rq_reqmsg, 1);
281 lit->opc = NTOH__u64((__u64)it->it_op);
283 /* pack the intended request */
284 mds_unlink_pack(req, 2, dir, NULL,
285 it->it_op == IT_UNLINK ? S_IFREG : S_IFDIR,
286 de->d_name.name, de->d_name.len);
288 req->rq_replen = lustre_msg_size(3, repsize);
289 } else if (it->it_op == IT_GETATTR || it->it_op == IT_RENAME ||
290 it->it_op == IT_OPEN || it->it_op == IT_SETATTR ||
291 it->it_op == IT_LOOKUP || it->it_op == IT_READLINK) {
292 size[2] = sizeof(struct mds_body);
293 size[3] = de->d_name.len + 1;
295 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
300 /* pack the intent */
301 lit = lustre_msg_buf(req->rq_reqmsg, 1);
302 lit->opc = NTOH__u64((__u64)it->it_op);
304 /* pack the intended request */
305 mds_getattr_pack(req, 2, dir, de->d_name.name, de->d_name.len);
307 /* get ready for the reply */
308 req->rq_replen = lustre_msg_size(3, repsize);
309 } else if (it->it_op == IT_READDIR) {
310 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 1,
315 /* get ready for the reply */
316 req->rq_replen = lustre_msg_size(1, repsize);
321 #warning FIXME: the data here needs to be different if a lock was granted for a different inode
322 rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id,
323 lock_type, NULL, 0, lock_mode, &flags,
324 ldlm_completion_ast, mdc_blocking_ast, data,
327 /* This can go when we're sure that this can never happen */
330 if (rc == ELDLM_LOCK_ABORTED) {
332 memset(lockh, 0, sizeof(*lockh));
334 } else if (rc != 0) {
335 CERROR("ldlm_cli_enqueue: %d\n", rc);
339 dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
340 it->it_disposition = (int) dlm_rep->lock_policy_res1;
341 it->it_status = (int) dlm_rep->lock_policy_res2;
342 it->it_lock_mode = lock_mode;
348 int mdc_open(struct lustre_handle *conn, obd_id ino, int type, int flags,
349 struct lov_stripe_md *smd, __u64 cookie, __u64 *fh,
350 struct ptlrpc_request **request)
352 struct mds_body *body;
353 int rc, size[2] = {sizeof(*body)}, bufcount = 1;
354 struct ptlrpc_request *req;
359 size[1] = smd->lmd_easize;
362 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_OPEN, bufcount, size,
365 GOTO(out, rc = -ENOMEM);
367 req->rq_flags |= PTL_RPC_FL_REPLAY;
368 body = lustre_msg_buf(req->rq_reqmsg, 0);
370 ll_ino2fid(&body->fid1, ino, 0, type);
371 body->flags = HTON__u32(flags);
372 body->extra = cookie;
375 lov_packmd(lustre_msg_buf(req->rq_reqmsg, 1), smd);
377 req->rq_replen = lustre_msg_size(1, size);
379 rc = ptlrpc_queue_wait(req);
380 rc = ptlrpc_check_status(req, rc);
382 body = lustre_msg_buf(req->rq_repmsg, 0);
383 mds_unpack_body(body);
393 int mdc_close(struct lustre_handle *conn,
394 obd_id ino, int type, __u64 fh, struct ptlrpc_request **request)
396 struct mds_body *body;
397 int rc, size = sizeof(*body);
398 struct ptlrpc_request *req;
400 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_CLOSE, 1, &size,
403 GOTO(out, rc = -ENOMEM);
405 body = lustre_msg_buf(req->rq_reqmsg, 0);
406 ll_ino2fid(&body->fid1, ino, 0, type);
409 req->rq_replen = lustre_msg_size(0, NULL);
411 rc = ptlrpc_queue_wait(req);
412 rc = ptlrpc_check_status(req, rc);
420 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
421 char *addr, struct ptlrpc_request **request)
423 struct ptlrpc_connection *connection =
424 client_conn2cli(conn)->cl_import.imp_connection;
425 struct ptlrpc_request *req = NULL;
426 struct ptlrpc_bulk_desc *desc = NULL;
427 struct ptlrpc_bulk_page *bulk = NULL;
428 struct mds_body *body;
429 int rc, size = sizeof(*body);
432 CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
434 desc = ptlrpc_prep_bulk(connection);
436 GOTO(out, rc = -ENOMEM);
438 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_READPAGE, 1, &size,
441 GOTO(out2, rc = -ENOMEM);
443 bulk = ptlrpc_prep_bulk_page(desc);
444 bulk->b_buflen = PAGE_SIZE;
446 bulk->b_xid = req->rq_xid;
447 desc->b_portal = MDS_BULK_PORTAL;
449 rc = ptlrpc_register_bulk(desc);
451 CERROR("couldn't setup bulk sink: error %d.\n", rc);
455 body = lustre_msg_buf(req->rq_reqmsg, 0);
457 body->fid1.f_type = type;
460 req->rq_replen = lustre_msg_size(1, &size);
461 rc = ptlrpc_queue_wait(req);
462 rc = ptlrpc_check_status(req, rc);
464 ptlrpc_abort_bulk(desc);
467 body = lustre_msg_buf(req->rq_repmsg, 0);
468 mds_unpack_body(body);
473 ptlrpc_free_bulk(desc);
479 int mdc_statfs(struct lustre_handle *conn, struct statfs *sfs,
480 struct ptlrpc_request **request)
482 struct obd_statfs *osfs;
483 struct ptlrpc_request *req;
484 int rc, size = sizeof(*osfs);
487 req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_STATFS, 0, NULL,
490 GOTO(out, rc = -ENOMEM);
491 req->rq_replen = lustre_msg_size(1, &size);
493 rc = ptlrpc_queue_wait(req);
494 rc = ptlrpc_check_status(req, rc);
499 osfs = lustre_msg_buf(req->rq_repmsg, 0);
500 obd_statfs_unpack(osfs, sfs);
509 struct obd_ops mdc_obd_ops = {
510 o_setup: client_obd_setup,
511 o_cleanup: client_obd_cleanup,
512 o_connect: client_obd_connect,
513 o_disconnect: client_obd_disconnect,
516 static int __init ptlrpc_request_init(void)
518 return class_register_type(&mdc_obd_ops, LUSTRE_MDC_NAME);
521 static void __exit ptlrpc_request_exit(void)
523 class_unregister_type(LUSTRE_MDC_NAME);
526 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
527 MODULE_DESCRIPTION("Lustre Metadata Client v1.0");
528 MODULE_LICENSE("GPL");
530 EXPORT_SYMBOL(mdc_getstatus);
531 EXPORT_SYMBOL(mdc_getlovinfo);
532 EXPORT_SYMBOL(mdc_enqueue);
533 EXPORT_SYMBOL(mdc_getattr);
534 EXPORT_SYMBOL(mdc_statfs);
535 EXPORT_SYMBOL(mdc_create);
536 EXPORT_SYMBOL(mdc_unlink);
537 EXPORT_SYMBOL(mdc_rename);
538 EXPORT_SYMBOL(mdc_link);
539 EXPORT_SYMBOL(mdc_readpage);
540 EXPORT_SYMBOL(mdc_setattr);
541 EXPORT_SYMBOL(mdc_close);
542 EXPORT_SYMBOL(mdc_open);
544 module_init(ptlrpc_request_init);
545 module_exit(ptlrpc_request_exit);