3 * Storage Target Handling functions
5 * Lustre Object Server Module (OST)
7 * Copyright (C) 2001 Cluster File Systems, Inc.
9 * This code is issued under the GNU General Public License.
10 * See the file COPYING in this distribution
12 * by Peter Braam <braam@clusterfs.com>
14 * This server is single threaded at present (but can easily be multi threaded).
15 * For testing and management it is treated as an obd_device, although it does
16 * not export a full OBD method table (the requests are coming in over the wire,
17 * so object target modules do not have a full method table.)
24 #include <linux/version.h>
25 #include <linux/module.h>
27 #include <linux/stat.h>
28 #include <linux/locks.h>
29 #include <linux/ext2_fs.h>
30 #include <linux/quotaops.h>
31 #include <asm/unistd.h>
32 #include <linux/obd_support.h>
33 #include <linux/obd.h>
34 #include <linux/obd_class.h>
35 #include <linux/lustre_lib.h>
36 #include <linux/lustre_idl.h>
37 #include <linux/lustre_mds.h>
38 #include <linux/obd_class.h>
41 static int ost_queue_req(struct obd_device *obddev, struct ptlrpc_request *req)
43 struct ptlrpc_request *srv_req;
44 struct ost_obd *ost = &obddev->u.ost;
51 srv_req = kmalloc(sizeof(*srv_req), GFP_KERNEL);
57 printk("---> OST at %d %p, incoming req %p, srv_req %p\n",
58 __LINE__, ost, req, srv_req);
60 memset(srv_req, 0, sizeof(*req));
62 /* move the request buffer */
63 srv_req->rq_reqbuf = req->rq_reqbuf;
64 srv_req->rq_reqlen = req->rq_reqlen;
65 srv_req->rq_ost = ost;
67 /* remember where it came from */
68 srv_req->rq_reply_handle = req;
70 list_add(&srv_req->rq_list, &ost->ost_reqs);
71 wake_up(&ost->ost_waitq);
76 /* XXX replace with networking code */
77 int ost_reply(struct obd_device *obddev, struct ptlrpc_request *req)
79 struct ptlrpc_request *clnt_req = req->rq_reply_handle;
82 printk("ost_reply: req %p clnt_req at %p\n", req, clnt_req);
84 /* free the request buffer */
85 kfree(req->rq_reqbuf);
86 req->rq_reqbuf = NULL;
88 /* move the reply to the client */
89 clnt_req->rq_replen = req->rq_replen;
90 clnt_req->rq_repbuf = req->rq_repbuf;
92 printk("---> client req %p repbuf %p len %d status %d\n",
93 clnt_req, clnt_req->rq_repbuf, clnt_req->rq_replen,
94 req->rq_rephdr->status);
96 req->rq_repbuf = NULL;
99 /* free the server request */
101 /* wake up the client */
102 wake_up_interruptible(&clnt_req->rq_wait_for_rep);
107 int ost_error(struct obd_device *obddev, struct ptlrpc_request *req)
109 struct ptlrep_hdr *hdr;
112 hdr = kmalloc(sizeof(*hdr), GFP_KERNEL);
118 memset(hdr, 0, sizeof(*hdr));
120 hdr->seqno = req->rq_reqhdr->seqno;
121 hdr->status = req->rq_status;
122 hdr->type = OST_TYPE_ERR;
124 req->rq_repbuf = (char *)hdr;
125 req->rq_replen = sizeof(*hdr);
128 return ost_reply(obddev, req);
131 static int ost_destroy(struct ost_obd *ost, struct ptlrpc_request *req)
133 struct obd_conn conn;
138 conn.oc_id = req->rq_req.ost->connid;
139 conn.oc_dev = ost->ost_tgt;
141 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
142 &req->rq_replen, &req->rq_repbuf);
144 printk("ost_destroy: cannot pack reply\n");
148 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_destroy
149 (&conn, &req->rq_req.ost->oa);
155 static int ost_getattr(struct ost_obd *ost, struct ptlrpc_request *req)
157 struct obd_conn conn;
161 printk("ost getattr entered\n");
163 conn.oc_id = req->rq_req.ost->connid;
164 conn.oc_dev = ost->ost_tgt;
166 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
167 &req->rq_replen, &req->rq_repbuf);
169 printk("ost_getattr: cannot pack reply\n");
172 req->rq_rep.ost->oa.o_id = req->rq_req.ost->oa.o_id;
173 req->rq_rep.ost->oa.o_valid = req->rq_req.ost->oa.o_valid;
175 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_getattr
176 (&conn, &req->rq_rep.ost->oa);
182 static int ost_create(struct ost_obd *ost, struct ptlrpc_request *req)
184 struct obd_conn conn;
189 conn.oc_id = req->rq_req.ost->connid;
190 conn.oc_dev = ost->ost_tgt;
192 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
193 &req->rq_replen, &req->rq_repbuf);
195 printk("ost_create: cannot pack reply\n");
199 memcpy(&req->rq_rep.ost->oa, &req->rq_req.ost->oa, sizeof(req->rq_req.ost->oa));
201 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_create
202 (&conn, &req->rq_rep.ost->oa);
209 static int ost_setattr(struct ost_obd *ost, struct ptlrpc_request *req)
211 struct obd_conn conn;
216 conn.oc_id = req->rq_req.ost->connid;
217 conn.oc_dev = ost->ost_tgt;
219 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
220 &req->rq_replen, &req->rq_repbuf);
222 printk("ost_setattr: cannot pack reply\n");
226 memcpy(&req->rq_rep.ost->oa, &req->rq_req.ost->oa, sizeof(req->rq_req.ost->oa));
228 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_setattr
229 (&conn, &req->rq_rep.ost->oa);
235 static int ost_connect(struct ost_obd *ost, struct ptlrpc_request *req)
237 struct obd_conn conn;
242 conn.oc_dev = ost->ost_tgt;
244 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
245 &req->rq_replen, &req->rq_repbuf);
247 printk("ost_setattr: cannot pack reply\n");
251 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_connect(&conn);
253 printk("ost_connect: rep buffer %p, id %d\n", req->rq_repbuf,
255 req->rq_rep.ost->connid = conn.oc_id;
261 static int ost_disconnect(struct ost_obd *ost, struct ptlrpc_request *req)
263 struct obd_conn conn;
268 conn.oc_dev = ost->ost_tgt;
269 conn.oc_id = req->rq_req.ost->connid;
271 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
272 &req->rq_replen, &req->rq_repbuf);
274 printk("ost_setattr: cannot pack reply\n");
278 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_disconnect(&conn);
284 static int ost_get_info(struct ost_obd *ost, struct ptlrpc_request *req)
286 struct obd_conn conn;
294 conn.oc_id = req->rq_req.ost->connid;
295 conn.oc_dev = ost->ost_tgt;
297 ptr = ost_req_buf1(req->rq_req.ost);
298 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_get_info
299 (&conn, req->rq_req.ost->buflen1, ptr, &vallen, &val);
301 rc = ost_pack_rep(val, vallen, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
302 &req->rq_replen, &req->rq_repbuf);
304 printk("ost_setattr: cannot pack reply\n");
314 static struct page * ext2_get_page(struct inode *dir, unsigned long n)
316 struct address_space *mapping = dir->i_mapping;
317 struct page *page = read_cache_page(mapping, n,
318 (filler_t*)mapping->a_ops->readpage, NULL);
322 if (!Page_Uptodate(page))
324 if (!PageChecked(page))
325 ext2_check_page(page);
333 return ERR_PTR(-EIO);
336 static inline void ext2_put_page(struct page *page)
339 page_cache_release(page);
342 /* Releases the page */
343 void ext2_set_link(struct inode *dir, struct ext2_dir_entry_2 *de,
344 struct page *page, struct inode *inode)
346 unsigned from = (char *) de - (char *) page_address(page);
347 unsigned to = from + le16_to_cpu(de->rec_len);
351 err = page->mapping->a_ops->prepare_write(NULL, page, from, to);
354 de->inode = cpu_to_le32(inode->i_ino);
355 ext2_set_de_type (de, inode);
356 dir->i_mtime = dir->i_ctime = CURRENT_TIME;
357 err = ext2_commit_chunk(page, from, to);
362 static int ext2_commit_chunk(struct page *page, unsigned from, unsigned to)
364 struct inode *dir = page->mapping->host;
366 dir->i_version = ++event;
367 SetPageUptodate(page);
368 set_page_clean(page);
370 //page->mapping->a_ops->commit_write(NULL, page, from, to);
372 // err = waitfor_one_page(page);
378 int ost_prepw(struct ost_obd *obddev, struct ptlrpc_request *req)
381 struct obd_conn conn;
387 struct obd_ioo **ioo;
391 tmp1 = ost_req_buf1(req);
392 tmp2 = ost_req_buf2(req);
393 objcount = req->buflen1 / sizeof(**ioo);
396 for (i=0 ; i<objcount ; i++) {
399 conn.oc_id = req->rq_req.ost->connid;
400 conn.oc_dev = ost->ost_tgt;
402 rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep,
403 &req->rq_replen, &req->rq_repbuf);
405 printk("ost_create: cannot pack reply\n");
409 memcpy(&req->rq_rep.ost->oa, &req->rq_req.ost->oa, sizeof(req->rq_req.ost->oa));
411 req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_create
412 (&conn, &req->rq_rep.ost->oa);
422 int ost_handle(struct obd_device *obddev, struct ptlrpc_request *req)
425 struct ost_obd *ost = &obddev->u.ost;
426 struct ptlreq_hdr *hdr;
429 printk("ost_handle: req at %p\n", req);
431 hdr = (struct ptlreq_hdr *)req->rq_reqbuf;
432 if (NTOH__u32(hdr->type) != OST_TYPE_REQ) {
433 printk("lustre_ost: wrong packet type sent %d\n",
434 NTOH__u32(hdr->type));
439 rc = ost_unpack_req(req->rq_reqbuf, req->rq_reqlen,
440 &req->rq_reqhdr, &req->rq_req.ost);
442 printk("lustre_ost: Invalid request\n");
447 switch (req->rq_reqhdr->opc) {
450 CDEBUG(D_INODE, "connect\n");
451 printk("----> connect \n");
452 rc = ost_connect(ost, req);
455 CDEBUG(D_INODE, "disconnect\n");
456 rc = ost_disconnect(ost, req);
459 CDEBUG(D_INODE, "get_info\n");
460 rc = ost_get_info(ost, req);
463 CDEBUG(D_INODE, "create\n");
464 rc = ost_create(ost, req);
467 CDEBUG(D_INODE, "destroy\n");
468 rc = ost_destroy(ost, req);
471 CDEBUG(D_INODE, "getattr\n");
472 rc = ost_getattr(ost, req);
475 CDEBUG(D_INODE, "setattr\n");
476 rc = ost_setattr(ost, req);
479 CDEBUG(D_INODE, "prepw\n");
480 rc = ost_prepw(ost, req);
483 req->rq_status = -ENOTSUPP;
484 return ost_error(obddev, req);
490 printk("ost: processing error %d\n", rc);
491 ost_error(obddev, req);
493 CDEBUG(D_INODE, "sending reply\n");
494 ost_reply(obddev, req);
500 int ost_main(void *arg)
502 struct obd_device *obddev = (struct obd_device *) arg;
503 struct ost_obd *ost = &obddev->u.ost;
505 printk("---> %d\n", __LINE__);
509 printk("---> %d\n", __LINE__);
511 printk("---> %d\n", __LINE__);
512 spin_lock_irq(¤t->sigmask_lock);
513 printk("---> %d\n", __LINE__);
514 sigfillset(¤t->blocked);
515 printk("---> %d\n", __LINE__);
516 recalc_sigpending(current);
517 printk("---> %d\n", __LINE__);
518 spin_unlock_irq(¤t->sigmask_lock);
519 printk("---> %d\n", __LINE__);
521 printk("---> %d\n", __LINE__);
522 sprintf(current->comm, "lustre_ost");
523 printk("---> %d\n", __LINE__);
525 /* Record that the thread is running */
526 ost->ost_thread = current;
527 printk("---> %d\n", __LINE__);
528 wake_up(&ost->ost_done_waitq);
529 printk("---> %d\n", __LINE__);
531 /* XXX maintain a list of all managed devices: insert here */
533 /* And now, wait forever for commit wakeup events. */
535 struct ptlrpc_request *request;
538 if (ost->ost_flags & OST_EXIT)
542 wake_up(&ost->ost_done_waitq);
543 interruptible_sleep_on(&ost->ost_waitq);
545 CDEBUG(D_INODE, "lustre_ost wakes\n");
546 CDEBUG(D_INODE, "pick up req here and continue\n");
548 if (list_empty(&ost->ost_reqs)) {
549 CDEBUG(D_INODE, "woke because of timer\n");
551 printk("---> %d\n", __LINE__);
552 request = list_entry(ost->ost_reqs.next,
553 struct ptlrpc_request, rq_list);
554 printk("---> %d\n", __LINE__);
555 list_del(&request->rq_list);
556 rc = ost_handle(obddev, request);
560 /* XXX maintain a list of all managed devices: cleanup here */
561 printk("---> %d\n", __LINE__);
562 ost->ost_thread = NULL;
563 printk("---> %d\n", __LINE__);
564 wake_up(&ost->ost_done_waitq);
565 printk("lustre_ost: exiting\n");
569 static void ost_stop_srv_thread(struct ost_obd *ost)
571 ost->ost_flags |= OST_EXIT;
573 while (ost->ost_thread) {
574 wake_up(&ost->ost_waitq);
575 sleep_on(&ost->ost_done_waitq);
579 static void ost_start_srv_thread(struct obd_device *obd)
581 struct ost_obd *ost = &obd->u.ost;
584 init_waitqueue_head(&ost->ost_waitq);
585 printk("---> %d\n", __LINE__);
586 init_waitqueue_head(&ost->ost_done_waitq);
587 printk("---> %d\n", __LINE__);
588 kernel_thread(ost_main, (void *)obd,
589 CLONE_VM | CLONE_FS | CLONE_FILES);
590 printk("---> %d\n", __LINE__);
591 while (!ost->ost_thread)
592 sleep_on(&ost->ost_done_waitq);
593 printk("---> %d\n", __LINE__);
597 /* mount the file system (secretly) */
598 static int ost_setup(struct obd_device *obddev, obd_count len,
602 struct obd_ioctl_data* data = buf;
603 struct ost_obd *ost = &obddev->u.ost;
604 struct obd_device *tgt;
608 if (data->ioc_dev < 0 || data->ioc_dev > MAX_OBD_DEVICES) {
613 tgt = &obd_dev[data->ioc_dev];
615 if ( ! (tgt->obd_flags & OBD_ATTACHED) ||
616 ! (tgt->obd_flags & OBD_SET_UP) ){
617 printk("device not attached or not set up (%d)\n",
623 ost->ost_conn.oc_dev = tgt;
624 err = tgt->obd_type->typ_ops->o_connect(&ost->ost_conn);
626 printk("lustre ost: fail to connect to device %d\n",
631 INIT_LIST_HEAD(&ost->ost_reqs);
632 ost->ost_thread = NULL;
635 spin_lock_init(&obddev->u.ost.ost_lock);
637 ost_start_srv_thread(obddev);
644 static int ost_cleanup(struct obd_device * obddev)
646 struct ost_obd *ost = &obddev->u.ost;
647 struct obd_device *tgt;
652 if ( !(obddev->obd_flags & OBD_SET_UP) ) {
657 if ( !list_empty(&obddev->obd_gen_clients) ) {
658 printk(KERN_WARNING __FUNCTION__ ": still has clients!\n");
663 ost_stop_srv_thread(ost);
665 if (!list_empty(&ost->ost_reqs)) {
666 // XXX reply with errors and clean up
667 CDEBUG(D_INODE, "Request list not empty!\n");
671 err = tgt->obd_type->typ_ops->o_disconnect(&ost->ost_conn);
673 printk("lustre ost: fail to disconnect device\n");
683 /* use obd ops to offer management infrastructure */
684 static struct obd_ops ost_obd_ops = {
686 o_cleanup: ost_cleanup,
689 static int __init ost_init(void)
691 obd_register_type(&ost_obd_ops, LUSTRE_OST_NAME);
695 static void __exit ost_exit(void)
697 obd_unregister_type(LUSTRE_OST_NAME);
700 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
701 MODULE_DESCRIPTION("Lustre Object Storage Target (OST) v0.01");
702 MODULE_LICENSE("GPL");
704 // for testing (maybe this stays)
705 EXPORT_SYMBOL(ost_queue_req);
707 module_init(ost_init);
708 module_exit(ost_exit);