- braindead first pass at a ptlbd backing store, it'll do for now

diff --git a/lustre/ptlbd/rpc.c b/lustre/ptlbd/rpc.c
index 7899436..1f8b90e 100644 (file)
--- a/lustre/ptlbd/rpc.c
+++ b/lustre/ptlbd/rpc.c
@@ -132,19 +132,34 @@ static int ptlbd_bulk_timeout(void *data)
         RETURN(1);
 }
 
-#define SILLY_MAX 2048
-static struct page *pages[SILLY_MAX] = {NULL,};
-
-static struct page * fake_page(int block_nr)
+void ptlbd_do_filp(struct file *filp, int op, struct ptlbd_niob *niobs, 
+                int page_count, struct list_head *page_list)
 {
-        if ( block_nr >= SILLY_MAX )
-                return NULL;
-
-        if (pages[block_nr] == NULL) {
-                void *vaddr = (void *)get_free_page(GFP_KERNEL);
-                pages[block_nr] = virt_to_page(vaddr);
-        } 
-        return pages[block_nr];
+        mm_segment_t old_fs;
+        struct list_head *pos;
+        ENTRY;
+
+        old_fs = get_fs();
+        set_fs(KERNEL_DS);
+
+        list_for_each(pos, page_list) {
+                ssize_t ret;
+                struct page *page = list_entry(pos, struct page, list);
+                loff_t offset = (niobs->n_block_nr << PAGE_SHIFT) + 
+                        niobs->n_offset;
+
+                if ( op == PTLBD_READ )
+                        ret = filp->f_op->read(filp, page_address(page), 
+                                        niobs->n_length, &offset);
+                else
+                        ret = filp->f_op->write(filp, page_address(page), 
+                                        niobs->n_length, &offset);
+
+                niobs++;
+        }
+
+        set_fs(old_fs);
+        EXIT;
 }
 
 int ptlbd_parse_req(struct ptlrpc_request *req)
@@ -153,8 +168,11 @@ int ptlbd_parse_req(struct ptlrpc_request *req)
         struct ptlbd_niob *niob, *niobs;
         struct ptlbd_rsp *rsp;
         struct ptlrpc_bulk_desc *desc;
+        struct file *filp = req->rq_obd->u.ptlbd.filp;
         struct l_wait_info lwi;
         int size[1], wait_flag, i, page_count, rc;
+        struct list_head *pos, *n;
+        LIST_HEAD(tmp_pages);
         ENTRY;
 
         rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
@@ -178,16 +196,21 @@ int ptlbd_parse_req(struct ptlrpc_request *req)
                 if (bulk == NULL)
                         GOTO(out_bulk, rc = -ENOMEM);
 
+                bulk->bp_page = alloc_page(GFP_KERNEL);
+                if (bulk->bp_page == NULL)
+                        GOTO(out_bulk, rc = -ENOMEM);
+                list_add(&bulk->bp_page->list, &tmp_pages);
+
                 /* 
                  * XXX what about the block number? 
                  */
                 bulk->bp_xid = niob->n_xid;
-                bulk->bp_page = fake_page(niob->n_block_nr);
                 bulk->bp_buf = page_address(bulk->bp_page);
                 bulk->bp_buflen = niob->n_length;
         }
 
         if ( op->op_cmd == PTLBD_READ ) {
+                ptlbd_do_filp(filp, PTLBD_READ, niobs, page_count, &tmp_pages);
                 rc = ptlrpc_bulk_put(desc);
                 wait_flag = PTL_BULK_FL_SENT;
         } else {
@@ -210,6 +233,8 @@ int ptlbd_parse_req(struct ptlrpc_request *req)
         rsp = lustre_msg_buf(req->rq_repmsg, 0);
         if ( rsp == NULL )
                 GOTO(out, rc = -EINVAL);
+        
+        ptlbd_do_filp(filp, PTLBD_WRITE, niobs, page_count, &tmp_pages);
 
         rsp->r_error_cnt = 42;
         rsp->r_status = 69;
@@ -218,6 +243,11 @@ int ptlbd_parse_req(struct ptlrpc_request *req)
         ptlrpc_reply(req->rq_svc, req);
 
 out_bulk:
+        list_for_each_safe(pos, n, &tmp_pages) {
+                struct page *page = list_entry(pos, struct page, list);
+                list_del(&page->list);
+                __free_page(page);
+        }
         ptlrpc_bulk_decref(desc);
 out:
         RETURN(rc);

diff --git a/lustre/ptlbd/server.c b/lustre/ptlbd/server.c
index 00c2376..d112069 100644 (file)
--- a/lustre/ptlbd/server.c
+++ b/lustre/ptlbd/server.c
@@ -31,80 +31,44 @@
 #include <linux/lprocfs_status.h>
 #include <linux/obd_ptlbd.h>
 
-#if 0
-static int ptlbd_sv_callback(struct ptlrpc_request *req)
-{
-        int rc;
-        ENTRY;
-
-        rc = ptlbd_parse_request(req);
-
-        rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
-        if ( rc )
-                GOTO(out, rc);
-
-        printk("callback got a friggin opc %d\n", req->rq_reqmsg->opc);
-
-out:
-        RETURN(rc);
-}
-#endif
-
 static int ptlbd_sv_already_setup = 1;
 
 static int ptlbd_sv_setup(struct obd_device *obddev, obd_count len, void *buf)
 {
-#if 0
-        struct obd_ioctl_data* data = buf;
-        struct obd_uuid server_uuid;
-#endif
         struct obd_uuid self_uuid = { "self" };
         struct ptlbd_obd *ptlbd = &obddev->u.ptlbd;
         int rc;
         ENTRY;
 
-#if 0
-        if (data->ioc_inllen1 < 1) {
-                CERROR("requires a PTLBD server UUID\n");
-                RETURN(rc = -EINVAL);
-        }
-
-        if (data->ioc_inllen1 > 37) {
-                CERROR("PTLBD server UUID must be less than 38 characters\n");
-                RETURN(rc = -EINVAL);
-        }
-
-        memcpy(server_uuid, data->ioc_inlbuf1, MIN(data->ioc_inllen1,
-                                                   sizeof(server_uuid)));
+        ptlbd->filp = filp_open("/tmp/ptlbd-backing-file-la-la-la", 
+                                        O_RDWR|O_CREAT, 0600);
+        if ( IS_ERR(ptlbd->filp) )
+                RETURN(PTR_ERR(ptlbd->filp));
 
-#endif
         ptlbd->ptlbd_service =
                 ptlrpc_init_svc(PTLBD_NEVENTS, PTLBD_NBUFS, PTLBD_BUFSIZE,
                                 PTLBD_MAXREQSIZE, PTLBD_REQUEST_PORTAL,
                                 PTLBD_REPLY_PORTAL, &self_uuid,
                                 ptlbd_parse_req, "ptlbd_sv");
 
-        if (!ptlbd->ptlbd_service) {
-                CERROR("failed to start service\n");
-                RETURN(rc = -ENOMEM);
-        }
+        if (ptlbd->ptlbd_service == NULL) 
+                GOTO(out_filp, rc = -ENOMEM);
 
         rc = ptlrpc_start_thread(obddev, ptlbd->ptlbd_service, "ptldb");
-        if (rc) {
-                CERROR("cannot start PTLBD thread: rc %d\n", rc);
-                LBUG();
+        if (rc != 0) 
                 GOTO(out_thread, rc);
-        }
 
         ptlbd_sv_already_setup = 1;
 
         RETURN(0);
 
- out_thread:
+out_thread:
         ptlrpc_stop_all_threads(ptlbd->ptlbd_service);
         ptlrpc_unregister_service(ptlbd->ptlbd_service);
+out_filp:
+        filp_close(ptlbd->filp, NULL);
 
-        return rc;
+        RETURN(rc);
 }
 
 static int ptlbd_sv_cleanup(struct obd_device *obddev)
@@ -116,29 +80,17 @@ static int ptlbd_sv_cleanup(struct obd_device *obddev)
 
         ptlrpc_stop_all_threads(ptlbd->ptlbd_service);
         ptlrpc_unregister_service(ptlbd->ptlbd_service);
+        if ( ! IS_ERR(ptlbd->filp) )
+                filp_close(ptlbd->filp, NULL);
 
         ptlbd_sv_already_setup = 0;
         RETURN(0);
 }
 
-#if 0
-static int ptlbd_sv_connect(struct lustre_handle *conn, struct obd_device *src,
-                        struct obd_uuid cluuid, struct recovd_obd *recovd,
-                        ptlrpc_recovery_cb_t recover)
-{
-        return class_connect(conn, src, cluuid);
-}
-#endif
-
 static struct obd_ops ptlbd_sv_obd_ops = {
         o_owner:        THIS_MODULE,
-/*        o_iocontrol:    ptlbd_iocontrol,*/
         o_setup:        ptlbd_sv_setup,
         o_cleanup:      ptlbd_sv_cleanup,
-#if 0
-        o_connect:      ptlbd_sv_connect,
-        o_disconnect:   class_disconnect
-#endif
 };
 
 int ptlbd_sv_init(void)