Whamcloud - gitweb
b=3031
[fs/lustre-release.git] / lustre / obdecho / echo.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
5  *   Author: Peter Braam <braam@clusterfs.com>
6  *   Author: Andreas Dilger <adilger@clusterfs.com>
7  *
8  *   This file is part of Lustre, http://www.lustre.org.
9  *
10  *   Lustre is free software; you can redistribute it and/or
11  *   modify it under the terms of version 2 of the GNU General Public
12  *   License as published by the Free Software Foundation.
13  *
14  *   Lustre is distributed in the hope that it will be useful,
15  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
16  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17  *   GNU General Public License for more details.
18  *
19  *   You should have received a copy of the GNU General Public License
20  *   along with Lustre; if not, write to the Free Software
21  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
22  */
23
24 #ifndef EXPORT_SYMTAB
25 # define EXPORT_SYMTAB
26 #endif
27
28 #include <linux/version.h>
29 #include <linux/module.h>
30 #include <linux/mm.h>
31 #include <linux/highmem.h>
32 #include <linux/fs.h>
33 #include <linux/stat.h>
34 #include <linux/sched.h>
35 #include <linux/smp_lock.h>
36 #include <linux/proc_fs.h>
37 #include <linux/init.h>
38 #include <asm/unistd.h>
39
40 #define DEBUG_SUBSYSTEM S_ECHO
41
42 #include <linux/obd_support.h>
43 #include <linux/obd_class.h>
44 #include <linux/obd_echo.h>
45 #include <linux/lustre_debug.h>
46 #include <linux/lustre_dlm.h>
47 #include <linux/lprocfs_status.h>
48
49 #define ECHO_INIT_OBJID      0x1000000000000000ULL
50 #define ECHO_HANDLE_MAGIC    0xabcd0123fedc9876ULL
51
52 #define ECHO_PERSISTENT_PAGES (ECHO_PERSISTENT_SIZE/PAGE_SIZE)
53 static struct page *echo_persistent_pages[ECHO_PERSISTENT_PAGES];
54
55 enum {
56         LPROC_ECHO_READ_BYTES = 1,
57         LPROC_ECHO_WRITE_BYTES = 2,
58         LPROC_ECHO_LAST
59 };
60
61 static int echo_connect(struct lustre_handle *conn, struct obd_device *obd,
62                         struct obd_uuid *cluuid, struct obd_connect_data *data,
63                         unsigned long connect_flags)
64 {
65         return class_connect(conn, obd, cluuid);
66 }
67
68 static int echo_disconnect(struct obd_export *exp, unsigned long flags)
69 {
70         unsigned long irqflags;
71
72         LASSERT (exp != NULL);
73
74         ldlm_cancel_locks_for_export(exp);
75
76         spin_lock_irqsave(&exp->exp_lock, irqflags);
77         exp->exp_flags = flags;
78         /* complete all outstanding replies */
79         spin_lock_irqsave(&exp->exp_lock, irqflags);
80         while (!list_empty(&exp->exp_outstanding_replies)) {
81                 struct ptlrpc_reply_state *rs =
82                         list_entry(exp->exp_outstanding_replies.next,
83                                    struct ptlrpc_reply_state, rs_exp_list);
84                 struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
85
86                 spin_lock(&svc->srv_lock);
87                 list_del_init(&rs->rs_exp_list);
88                 ptlrpc_schedule_difficult_reply(rs);
89                 spin_unlock(&svc->srv_lock);
90         }
91         spin_unlock_irqrestore(&exp->exp_lock, irqflags);
92
93         return class_disconnect(exp, flags);
94 }
95
96 static int echo_destroy_export(struct obd_export *exp)
97 {
98         ENTRY;
99         
100         target_destroy_export(exp);
101
102         RETURN(0);
103 }
104
105  static __u64 echo_next_id(struct obd_device *obddev)
106 {
107         obd_id id;
108
109         spin_lock(&obddev->u.echo.eo_lock);
110         id = ++obddev->u.echo.eo_lastino;
111         spin_unlock(&obddev->u.echo.eo_lock);
112
113         return id;
114 }
115
116 int echo_create(struct obd_export *exp, struct obdo *oa,
117                 struct lov_stripe_md **ea, struct obd_trans_info *oti)
118 {
119         struct obd_device *obd = class_exp2obd(exp);
120
121         if (!obd) {
122                 CERROR("invalid client cookie "LPX64"\n", 
123                        exp->exp_handle.h_cookie);
124                 return -EINVAL;
125         }
126
127         if (!(oa->o_mode && S_IFMT)) {
128                 CERROR("echo obd: no type!\n");
129                 return -ENOENT;
130         }
131
132         if (!(oa->o_valid & OBD_MD_FLTYPE)) {
133                 CERROR("invalid o_valid "LPX64"\n", oa->o_valid);
134                 return -EINVAL;
135         }
136
137         oa->o_id = echo_next_id(obd);
138         oa->o_valid = OBD_MD_FLID;
139
140         return 0;
141 }
142
143 int echo_destroy(struct obd_export *exp, struct obdo *oa,
144                  struct lov_stripe_md *ea, struct obd_trans_info *oti)
145 {
146         struct obd_device *obd = class_exp2obd(exp);
147
148         if (!obd) {
149                 CERROR("invalid client cookie "LPX64"\n", 
150                        exp->exp_handle.h_cookie);
151                 RETURN(-EINVAL);
152         }
153
154         if (!(oa->o_valid & OBD_MD_FLID)) {
155                 CERROR("obdo missing FLID valid flag: "LPX64"\n", oa->o_valid);
156                 RETURN(-EINVAL);
157         }
158
159         if (oa->o_id > obd->u.echo.eo_lastino || oa->o_id < ECHO_INIT_OBJID) {
160                 CERROR("bad destroy objid: "LPX64"\n", oa->o_id);
161                 RETURN(-EINVAL);
162         }
163
164
165         return 0;
166 }
167
168 static int echo_getattr(struct obd_export *exp, struct obdo *oa,
169                         struct lov_stripe_md *md)
170 {
171         struct obd_device *obd = class_exp2obd(exp);
172         obd_id id = oa->o_id;
173
174         if (!obd) {
175                 CERROR("invalid client cookie "LPX64"\n", 
176                        exp->exp_handle.h_cookie);
177                 RETURN(-EINVAL);
178         }
179
180         if (!(oa->o_valid & OBD_MD_FLID)) {
181                 CERROR("obdo missing FLID valid flag: "LPX64"\n", oa->o_valid);
182                 RETURN(-EINVAL);
183         }
184
185         obdo_cpy_md(oa, &obd->u.echo.eo_oa, oa->o_valid);
186         oa->o_id = id;
187
188         return 0;
189 }
190
191 static int echo_setattr(struct obd_export *exp, struct obdo *oa,
192                         struct lov_stripe_md *md, struct obd_trans_info *oti)
193 {
194         struct obd_device *obd = class_exp2obd(exp);
195
196         if (!obd) {
197                 CERROR("invalid client cookie "LPX64"\n", 
198                        exp->exp_handle.h_cookie);
199                 RETURN(-EINVAL);
200         }
201
202         if (!(oa->o_valid & OBD_MD_FLID)) {
203                 CERROR("obdo missing FLID valid flag: "LPX64"\n", oa->o_valid);
204                 RETURN(-EINVAL);
205         }
206
207         memcpy(&obd->u.echo.eo_oa, oa, sizeof(*oa));
208
209         if (oa->o_id & 4) {
210                 /* Save lock to force ACKed reply */
211                 ldlm_lock_addref (&obd->u.echo.eo_nl_lock, LCK_NL);
212                 oti->oti_ack_locks[0].mode = LCK_NL;
213                 oti->oti_ack_locks[0].lock = obd->u.echo.eo_nl_lock;
214         }
215         return 0;
216 }
217
218 static void
219 echo_page_debug_setup(struct page *page, int rw, obd_id id,
220                       __u64 offset, int len)
221 {
222         int   page_offset = offset & (PAGE_SIZE - 1);
223         char *addr        = ((char *)kmap(page)) + page_offset;
224
225         if (len % OBD_ECHO_BLOCK_SIZE != 0)
226                 CERROR("Unexpected block size %d\n", len);
227
228         while (len > 0) {
229                 if (rw & OBD_BRW_READ)
230                         block_debug_setup(addr, OBD_ECHO_BLOCK_SIZE,
231                                           offset, id);
232                 else
233                         block_debug_setup(addr, OBD_ECHO_BLOCK_SIZE,
234                                           0xecc0ecc0ecc0ecc0ULL,
235                                           0xecc0ecc0ecc0ecc0ULL);
236                 
237                 addr   += OBD_ECHO_BLOCK_SIZE;
238                 offset += OBD_ECHO_BLOCK_SIZE;
239                 len    -= OBD_ECHO_BLOCK_SIZE;
240         }
241
242         kunmap(page);
243 }
244
245 static int
246 echo_page_debug_check(struct page *page, obd_id id,
247                       __u64 offset, int len)
248 {
249         int   page_offset = offset & (PAGE_SIZE - 1);
250         char *addr        = ((char *)kmap(page)) + page_offset;
251         int   rc          = 0;
252         int   rc2;
253
254         if (len % OBD_ECHO_BLOCK_SIZE != 0)
255                 CERROR("Unexpected block size %d\n", len);
256
257         while (len > 0) {
258                 rc2 = block_debug_check("echo", addr, OBD_ECHO_BLOCK_SIZE,
259                                         offset, id);
260
261                 if (rc2 != 0 && rc == 0)
262                         rc = rc2;
263                 
264                 addr   += OBD_ECHO_BLOCK_SIZE;
265                 offset += OBD_ECHO_BLOCK_SIZE;
266                 len    -= OBD_ECHO_BLOCK_SIZE;
267         }
268
269         kunmap(page);
270
271         return (rc);
272 }
273
274 /* This allows us to verify that desc_private is passed unmolested */
275 #define DESC_PRIV 0x10293847
276
277 int echo_preprw(int cmd, struct obd_export *export, struct obdo *oa,
278                 int objcount, struct obd_ioobj *obj, int niocount,
279                 struct niobuf_remote *nb, struct niobuf_local *res,
280                 struct obd_trans_info *oti)
281 {
282         struct obd_device *obd;
283         struct niobuf_local *r = res;
284         int tot_bytes = 0;
285         int rc = 0;
286         int i;
287         ENTRY;
288
289         obd = export->exp_obd;
290         if (obd == NULL)
291                 RETURN(-EINVAL);
292
293         /* Temp fix to stop falling foul of osc_announce_cached() */
294         oa->o_valid &= ~(OBD_MD_FLBLOCKS | OBD_MD_FLGRANT);
295
296         memset(res, 0, sizeof(*res) * niocount);
297
298         CDEBUG(D_PAGE, "%s %d obdos with %d IOs\n",
299                cmd == OBD_BRW_READ ? "reading" : "writing", objcount, niocount);
300
301         if (oti)
302                 oti->oti_handle = (void *)DESC_PRIV;
303
304         for (i = 0; i < objcount; i++, obj++) {
305                 int gfp_mask = (obj->ioo_id & 1) ? GFP_HIGHUSER : GFP_KERNEL;
306                 int ispersistent = obj->ioo_id == ECHO_PERSISTENT_OBJID;
307                 int debug_setup = (!ispersistent &&
308                                    (oa->o_valid & OBD_MD_FLFLAGS) != 0 &&
309                                    (oa->o_flags & OBD_FL_DEBUG_CHECK) != 0);
310                 int j;
311
312                 for (j = 0 ; j < obj->ioo_bufcnt ; j++, nb++, r++) {
313
314                         if (ispersistent &&
315                             (nb->offset >> PAGE_SHIFT) < ECHO_PERSISTENT_PAGES) {
316                                 r->page = echo_persistent_pages[nb->offset >>
317                                                                 PAGE_SHIFT];
318                                 /* Take extra ref so __free_pages() can be called OK */
319                                 get_page (r->page);
320                         } else {
321                                 r->page = alloc_pages(gfp_mask, 0);
322                                 if (r->page == NULL) {
323                                         CERROR("can't get page %u/%u for id "
324                                                LPU64"\n",
325                                                j, obj->ioo_bufcnt, obj->ioo_id);
326                                         GOTO(preprw_cleanup, rc = -ENOMEM);
327                                 }
328                         }
329
330                         tot_bytes += nb->len;
331
332                         atomic_inc(&obd->u.echo.eo_prep);
333
334                         r->offset = nb->offset;
335                         r->len = nb->len;
336                         LASSERT((r->offset & ~PAGE_MASK) + r->len <= PAGE_SIZE);
337
338                         CDEBUG(D_PAGE, "$$$$ get page %p @ "LPU64" for %d\n",
339                                r->page, r->offset, r->len);
340
341                         if (cmd & OBD_BRW_READ)
342                                 r->rc = r->len;
343
344                         if (debug_setup)
345                                 echo_page_debug_setup(r->page, cmd, obj->ioo_id,
346                                                       r->offset, r->len);
347                 }
348         }
349         if (cmd & OBD_BRW_READ)
350                 lprocfs_counter_add(obd->obd_stats, LPROC_ECHO_READ_BYTES,
351                                     tot_bytes);
352         else
353                 lprocfs_counter_add(obd->obd_stats, LPROC_ECHO_WRITE_BYTES,
354                                     tot_bytes);
355
356         CDEBUG(D_PAGE, "%d pages allocated after prep\n",
357                atomic_read(&obd->u.echo.eo_prep));
358
359         RETURN(0);
360
361 preprw_cleanup:
362         /* It is possible that we would rather handle errors by  allow
363          * any already-set-up pages to complete, rather than tearing them
364          * all down again.  I believe that this is what the in-kernel
365          * prep/commit operations do.
366          */
367         CERROR("cleaning up %ld pages (%d obdos)\n", (long)(r - res), objcount);
368         while (r-- > res) {
369                 kunmap(r->page);
370                 /* NB if this is a persistent page, __free_pages will just
371                  * lose the extra ref gained above */
372                 __free_pages(r->page, 0);
373                 atomic_dec(&obd->u.echo.eo_prep);
374         }
375         memset(res, 0, sizeof(*res) * niocount);
376
377         return rc;
378 }
379
380 int echo_commitrw(int cmd, struct obd_export *export, struct obdo *oa,
381                   int objcount, struct obd_ioobj *obj, int niocount,
382                   struct niobuf_local *res, struct obd_trans_info *oti, int rc)
383 {
384         struct obd_device *obd;
385         struct niobuf_local *r = res;
386         int i, vrc = 0;
387         ENTRY;
388
389         obd = export->exp_obd;
390         if (obd == NULL)
391                 RETURN(-EINVAL);
392
393         if (rc)
394                 GOTO(commitrw_cleanup, rc);
395
396         if ((cmd & OBD_BRW_RWMASK) == OBD_BRW_READ) {
397                 CDEBUG(D_PAGE, "reading %d obdos with %d IOs\n",
398                        objcount, niocount);
399         } else {
400                 CDEBUG(D_PAGE, "writing %d obdos with %d IOs\n",
401                        objcount, niocount);
402         }
403
404         if (niocount && !r) {
405                 CERROR("NULL res niobuf with niocount %d\n", niocount);
406                 RETURN(-EINVAL);
407         }
408
409         LASSERT(oti == NULL || oti->oti_handle == (void *)DESC_PRIV);
410
411         for (i = 0; i < objcount; i++, obj++) {
412                 int verify = (rc == 0 &&
413                               obj->ioo_id != ECHO_PERSISTENT_OBJID &&
414                               (oa->o_valid & OBD_MD_FLFLAGS) != 0 &&
415                               (oa->o_flags & OBD_FL_DEBUG_CHECK) != 0);
416  
417                 int j;
418
419                 for (j = 0 ; j < obj->ioo_bufcnt ; j++, r++) {
420                         struct page *page = r->page;
421                         void *addr;
422
423                         if (page == NULL) {
424                                 CERROR("null page objid "LPU64":%p, buf %d/%d\n",
425                                        obj->ioo_id, page, j, obj->ioo_bufcnt);
426                                 GOTO(commitrw_cleanup, rc = -EFAULT);
427                         }
428
429                         addr = kmap(page);
430
431                         CDEBUG(D_PAGE, "$$$$ use page %p, addr %p@"LPU64"\n",
432                                r->page, addr, r->offset);
433
434                         if (verify) {
435                                 vrc = echo_page_debug_check(page, obj->ioo_id, 
436                                                             r->offset, r->len);
437                                 /* check all the pages always */
438                                 if (vrc != 0 && rc == 0)
439                                         rc = vrc;
440                         }
441
442                         kunmap(page);
443                         /* NB see comment above regarding persistent pages */
444                         __free_pages(page, 0);
445                         atomic_dec(&obd->u.echo.eo_prep);
446                 }
447         }
448         CDEBUG(D_PAGE, "%d pages remain after commit\n",
449                atomic_read(&obd->u.echo.eo_prep));
450         RETURN(rc);
451
452 commitrw_cleanup:
453         CERROR("cleaning up %ld pages (%d obdos)\n",
454                niocount - (long)(r - res) - 1, objcount);
455         while (++r < res + niocount) {
456                 struct page *page = r->page;
457
458                 /* NB see comment above regarding persistent pages */
459                 __free_pages(page, 0);
460                 atomic_dec(&obd->u.echo.eo_prep);
461         }
462         return rc;
463 }
464
465 static int echo_setup(struct obd_device *obd, obd_count len, void *buf)
466 {
467         int                        rc;
468         int                        lock_flags = 0;
469         struct ldlm_res_id         res_id = {.name = {1}};
470
471         ENTRY;
472
473         spin_lock_init(&obd->u.echo.eo_lock);
474         obd->u.echo.eo_lastino = ECHO_INIT_OBJID;
475
476         obd->obd_namespace = ldlm_namespace_new("echo-tgt",
477                                                 LDLM_NAMESPACE_SERVER);
478         if (obd->obd_namespace == NULL) {
479                 LBUG();
480                 RETURN(-ENOMEM);
481         }
482         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, res_id,
483                               LDLM_PLAIN, NULL, LCK_NL, &lock_flags,
484                               NULL, ldlm_completion_ast, NULL, NULL,
485                               NULL, 0, NULL, &obd->u.echo.eo_nl_lock);
486         LASSERT (rc == ELDLM_OK);
487
488
489
490         ptlrpc_init_client (LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
491                             "echo_ldlm_cb_client", &obd->obd_ldlm_client);
492         RETURN(0);
493 }
494
495 static int echo_cleanup(struct obd_device *obd, int flags)
496 {
497         int leaked;
498         ENTRY;
499
500         ldlm_lock_decref (&obd->u.echo.eo_nl_lock, LCK_NL);
501
502         /* XXX Bug 3413; wait for a bit to ensure the BL callback has
503          * happened before calling ldlm_namespace_free() */
504         set_current_state (TASK_UNINTERRUPTIBLE);
505         schedule_timeout (HZ);
506
507         ldlm_namespace_free(obd->obd_namespace, flags & OBD_OPT_FORCE);
508
509         leaked = atomic_read(&obd->u.echo.eo_prep);
510         if (leaked != 0)
511                 CERROR("%d prep/commitrw pages leaked\n", leaked);
512
513         RETURN(0);
514 }
515
516 int echo_attach(struct obd_device *obd, obd_count len, void *data)
517 {
518         struct lprocfs_static_vars lvars;
519         int rc;
520
521         lprocfs_init_vars(echo, &lvars);
522         rc = lprocfs_obd_attach(obd, lvars.obd_vars);
523         if (rc != 0)
524                 return rc;
525         rc = lprocfs_alloc_obd_stats(obd, LPROC_ECHO_LAST);
526         if (rc != 0)
527                 return rc;
528
529         lprocfs_counter_init(obd->obd_stats, LPROC_ECHO_READ_BYTES,
530                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
531         lprocfs_counter_init(obd->obd_stats, LPROC_ECHO_WRITE_BYTES,
532                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
533         return rc;
534 }
535
536 int echo_detach(struct obd_device *dev)
537 {
538         lprocfs_free_obd_stats(dev);
539         return lprocfs_obd_detach(dev);
540 }
541
542 static struct obd_ops echo_obd_ops = {
543         .o_owner           = THIS_MODULE,
544         .o_attach          = echo_attach,
545         .o_detach          = echo_detach,
546         .o_connect         = echo_connect,
547         .o_disconnect      = echo_disconnect,
548         .o_destroy_export  = echo_destroy_export,
549         .o_create          = echo_create,
550         .o_destroy         = echo_destroy,
551         .o_getattr         = echo_getattr,
552         .o_setattr         = echo_setattr,
553         .o_preprw          = echo_preprw,
554         .o_commitrw        = echo_commitrw,
555         .o_setup           = echo_setup,
556         .o_cleanup         = echo_cleanup
557 };
558
559 extern int echo_client_init(void);
560 extern void echo_client_exit(void);
561
562 static void
563 echo_persistent_pages_fini (void)
564 {
565         int     i;
566
567         for (i = 0; i < ECHO_PERSISTENT_PAGES; i++)
568                 if (echo_persistent_pages[i] != NULL) {
569                         __free_pages (echo_persistent_pages[i], 0);
570                         echo_persistent_pages[i] = NULL;
571                 }
572 }
573
574 static int
575 echo_persistent_pages_init (void)
576 {
577         struct page *pg;
578         int          i;
579
580         for (i = 0; i < ECHO_PERSISTENT_PAGES; i++) {
581                 int gfp_mask = (i < ECHO_PERSISTENT_PAGES/2) ?
582                         GFP_KERNEL : GFP_HIGHUSER;
583
584                 pg = alloc_pages (gfp_mask, 0);
585                 if (pg == NULL) {
586                         echo_persistent_pages_fini ();
587                         return (-ENOMEM);
588                 }
589
590                 memset (kmap (pg), 0, PAGE_SIZE);
591                 kunmap (pg);
592
593                 echo_persistent_pages[i] = pg;
594         }
595
596         return (0);
597 }
598
599 static int __init obdecho_init(void)
600 {
601         struct lprocfs_static_vars lvars;
602         int rc;
603
604         printk(KERN_INFO "Lustre: Echo OBD driver; info@clusterfs.com\n");
605
606         LASSERT(PAGE_SIZE % OBD_ECHO_BLOCK_SIZE == 0);
607
608         lprocfs_init_vars(echo, &lvars);
609
610         rc = echo_persistent_pages_init ();
611         if (rc != 0)
612                 goto failed_0;
613
614         rc = class_register_type(&echo_obd_ops, NULL, lvars.module_vars,
615                                  OBD_ECHO_DEVICENAME);
616         if (rc != 0)
617                 goto failed_1;
618
619         rc = echo_client_init();
620         if (rc == 0)
621                 RETURN (0);
622
623         class_unregister_type(OBD_ECHO_DEVICENAME);
624  failed_1:
625         echo_persistent_pages_fini ();
626  failed_0:
627         RETURN(rc);
628 }
629
630 static void /*__exit*/ obdecho_exit(void)
631 {
632         echo_client_exit();
633         class_unregister_type(OBD_ECHO_DEVICENAME);
634         echo_persistent_pages_fini ();
635 }
636
637 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
638 MODULE_DESCRIPTION("Lustre Testing Echo OBD driver");
639 MODULE_LICENSE("GPL");
640
641 module_init(obdecho_init);
642 module_exit(obdecho_exit);