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 #define EXPORT_SYMTAB
  25
  26 #include <linux/version.h>
  27 #include <linux/module.h>
  28 #include <linux/mm.h>
  29 #include <linux/highmem.h>
  30 #include <linux/fs.h>
  31 #include <linux/stat.h>
  32 #include <linux/sched.h>
  33 #include <linux/smp_lock.h>
  34 #include <linux/ext2_fs.h>
  35 #include <linux/quotaops.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_OBJECT0_NPAGES  16
  53 static struct page *echo_object0_pages[ECHO_OBJECT0_NPAGES];
  54
  55 enum {
  56         LPROC_ECHO_READ_BYTES = 1,
  57         LPROC_ECHO_WRITE_BYTES = 2,
  58         LPROC_ECHO_LAST = LPROC_ECHO_WRITE_BYTES +1
  59 };
  60
  61 static int echo_connect(struct lustre_handle *conn, struct obd_device *obd,
  62                         struct obd_uuid *cluuid)
  63 {
  64         return class_connect(conn, obd, cluuid);
  65 }
  66
  67 static int echo_disconnect(struct lustre_handle *conn, int flags)
  68 {
  69         struct obd_export *exp = class_conn2export(conn);
  70
  71         LASSERT (exp != NULL);
  72
  73         ldlm_cancel_locks_for_export(exp);
  74         class_export_put(exp);
  75         return class_disconnect(conn, flags);
  76 }
  77
  78 static __u64 echo_next_id(struct obd_device *obddev)
  79 {
  80         obd_id id;
  81
  82         spin_lock(&obddev->u.echo.eo_lock);
  83         id = ++obddev->u.echo.eo_lastino;
  84         spin_unlock(&obddev->u.echo.eo_lock);
  85
  86         return id;
  87 }
  88
  89 int echo_create(struct lustre_handle *conn, struct obdo *oa,
  90                 struct lov_stripe_md **ea, struct obd_trans_info *oti)
  91 {
  92         struct obd_device *obd = class_conn2obd(conn);
  93
  94         if (!obd) {
  95                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
  96                 return -EINVAL;
  97         }
  98
  99         if (!(oa->o_mode && S_IFMT)) {
 100                 CERROR("echo obd: no type!\n");
 101                 return -ENOENT;
 102         }
 103
 104         if (!(oa->o_valid & OBD_MD_FLTYPE)) {
 105                 CERROR("invalid o_valid %08x\n", oa->o_valid);
 106                 return -EINVAL;
 107         }
 108
 109         oa->o_id = echo_next_id(obd);
 110         oa->o_valid = OBD_MD_FLID;
 111         atomic_inc(&obd->u.echo.eo_create);
 112
 113         return 0;
 114 }
 115
 116 int echo_destroy(struct lustre_handle *conn, struct obdo *oa,
 117                  struct lov_stripe_md *ea, struct obd_trans_info *oti)
 118 {
 119         struct obd_device *obd = class_conn2obd(conn);
 120
 121         if (!obd) {
 122                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
 123                 RETURN(-EINVAL);
 124         }
 125
 126         if (!(oa->o_valid & OBD_MD_FLID)) {
 127                 CERROR("obdo missing FLID valid flag: %08x\n", oa->o_valid);
 128                 RETURN(-EINVAL);
 129         }
 130
 131         if (oa->o_id > obd->u.echo.eo_lastino || oa->o_id < ECHO_INIT_OBJID) {
 132                 CERROR("bad destroy objid: "LPX64"\n", oa->o_id);
 133                 RETURN(-EINVAL);
 134         }
 135
 136         atomic_inc(&obd->u.echo.eo_destroy);
 137
 138         return 0;
 139 }
 140
 141 static int echo_open(struct lustre_handle *conn, struct obdo *oa,
 142                      struct lov_stripe_md *md, struct obd_trans_info *oti,
 143                      struct obd_client_handle *och)
 144 {
 145         struct lustre_handle *fh = obdo_handle (oa);
 146         struct obd_device    *obd = class_conn2obd (conn);
 147
 148         if (!obd) {
 149                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
 150                 return (-EINVAL);
 151         }
 152
 153         if (!(oa->o_valid & OBD_MD_FLID)) {
 154                 CERROR ("obdo missing FLID valid flag: %08x\n", oa->o_valid);
 155                 return (-EINVAL);
 156         }
 157
 158         fh->cookie = ECHO_HANDLE_MAGIC;
 159
 160         oa->o_valid |= OBD_MD_FLHANDLE;
 161         return 0;
 162 }
 163
 164 static int echo_close(struct lustre_handle *conn, struct obdo *oa,
 165                       struct lov_stripe_md *md, struct obd_trans_info *oti)
 166 {
 167         struct lustre_handle *fh = obdo_handle (oa);
 168         struct obd_device    *obd = class_conn2obd(conn);
 169
 170         if (!obd) {
 171                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
 172                 return (-EINVAL);
 173         }
 174
 175         if (!(oa->o_valid & OBD_MD_FLHANDLE)) {
 176                 CERROR("obdo missing FLHANDLE valid flag: %08x\n", oa->o_valid);
 177                 return (-EINVAL);
 178         }
 179
 180         if (fh->cookie != ECHO_HANDLE_MAGIC) {
 181                 CERROR ("invalid file handle on close: "LPX64"\n", fh->cookie);
 182                 return (-EINVAL);
 183         }
 184
 185         return 0;
 186 }
 187
 188 static int echo_getattr(struct lustre_handle *conn, struct obdo *oa,
 189                         struct lov_stripe_md *md)
 190 {
 191         struct obd_device *obd = class_conn2obd(conn);
 192         obd_id id = oa->o_id;
 193
 194         if (!obd) {
 195                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
 196                 RETURN(-EINVAL);
 197         }
 198
 199         if (!(oa->o_valid & OBD_MD_FLID)) {
 200                 CERROR("obdo missing FLID valid flag: %08x\n", oa->o_valid);
 201                 RETURN(-EINVAL);
 202         }
 203
 204         obdo_cpy_md(oa, &obd->u.echo.oa, oa->o_valid);
 205         oa->o_id = id;
 206
 207         return 0;
 208 }
 209
 210 static int echo_setattr(struct lustre_handle *conn, struct obdo *oa,
 211                         struct lov_stripe_md *md, struct obd_trans_info *oti)
 212 {
 213         struct obd_device *obd = class_conn2obd(conn);
 214
 215         if (!obd) {
 216                 CERROR("invalid client cookie "LPX64"\n", conn->cookie);
 217                 RETURN(-EINVAL);
 218         }
 219
 220         if (!(oa->o_valid & OBD_MD_FLID)) {
 221                 CERROR("obdo missing FLID valid flag: %08x\n", oa->o_valid);
 222                 RETURN(-EINVAL);
 223         }
 224
 225         memcpy(&obd->u.echo.oa, oa, sizeof(*oa));
 226
 227         atomic_inc(&obd->u.echo.eo_setattr);
 228
 229         return 0;
 230 }
 231
 232 /* This allows us to verify that desc_private is passed unmolested */
 233 #define DESC_PRIV 0x10293847
 234
 235 int echo_preprw(int cmd, struct obd_export *export, struct obdo *oa,
 236                 int objcount, struct obd_ioobj *obj, int niocount,
 237                 struct niobuf_remote *nb, struct niobuf_local *res,
 238                 struct obd_trans_info *oti)
 239 {
 240         struct obd_device *obd;
 241         struct niobuf_local *r = res;
 242         int tot_bytes = 0;
 243         int rc = 0;
 244         int i;
 245         ENTRY;
 246
 247         obd = export->exp_obd;
 248         if (obd == NULL)
 249                 RETURN(-EINVAL);
 250
 251         /* Temp fix to stop falling foul of osc_announce_cached() */
 252         oa->o_valid &= ~(OBD_MD_FLBLOCKS | OBD_MD_FLRDEV);
 253
 254         memset(res, 0, sizeof(*res) * niocount);
 255
 256         CDEBUG(D_PAGE, "%s %d obdos with %d IOs\n",
 257                cmd == OBD_BRW_READ ? "reading" : "writing", objcount, niocount);
 258
 259         if (oti)
 260                 oti->oti_handle = (void *)DESC_PRIV;
 261
 262         for (i = 0; i < objcount; i++, obj++) {
 263                 int gfp_mask = (obj->ioo_id & 1) ? GFP_HIGHUSER : GFP_KERNEL;
 264                 int isobj0 = obj->ioo_id == 0;
 265                 int verify = !isobj0;
 266                 int j;
 267
 268                 for (j = 0 ; j < obj->ioo_bufcnt ; j++, nb++, r++) {
 269
 270                         if (isobj0 &&
 271                             (nb->offset >> PAGE_SHIFT) < ECHO_OBJECT0_NPAGES) {
 272                                 r->page = echo_object0_pages[nb->offset >>
 273                                                              PAGE_SHIFT];
 274                                 /* Take extra ref so __free_pages() can be called OK */
 275                                 get_page (r->page);
 276                         } else {
 277                                 r->page = alloc_pages(gfp_mask, 0);
 278                                 if (r->page == NULL) {
 279                                         CERROR("can't get page %u/%u for id "
 280                                                LPU64"\n",
 281                                                j, obj->ioo_bufcnt, obj->ioo_id);
 282                                         GOTO(preprw_cleanup, rc = -ENOMEM);
 283                                 }
 284                         }
 285
 286                         tot_bytes += nb->len;
 287
 288                         atomic_inc(&obd->u.echo.eo_prep);
 289
 290                         r->offset = nb->offset;
 291                         r->len = nb->len;
 292                         LASSERT((r->offset & ~PAGE_MASK) + r->len <= PAGE_SIZE);
 293
 294                         CDEBUG(D_PAGE, "$$$$ get page %p @ "LPU64" for %d\n",
 295                                r->page, r->offset, r->len);
 296
 297                         if (cmd & OBD_BRW_READ) {
 298                                 r->rc = r->len;
 299                                 if (verify) {
 300                                         page_debug_setup(kmap (r->page), r->len,
 301                                                          r->offset,obj->ioo_id);
 302                                         kunmap (r->page);
 303                                 }
 304                                 r->rc = r->len;
 305                         } else {
 306                                 if (verify) {
 307                                         page_debug_setup(kmap (r->page), r->len,
 308                                                          0xecc0ecc0ecc0ecc0,
 309                                                          0xecc0ecc0ecc0ecc0);
 310                                         kunmap (r->page);
 311                                 }
 312                         }
 313                 }
 314         }
 315         if (cmd & OBD_BRW_READ)
 316                 lprocfs_counter_add(obd->obd_stats, LPROC_ECHO_READ_BYTES,
 317                                     tot_bytes);
 318         else
 319                 lprocfs_counter_add(obd->obd_stats, LPROC_ECHO_WRITE_BYTES,
 320                                     tot_bytes);
 321
 322         CDEBUG(D_PAGE, "%d pages allocated after prep\n",
 323                atomic_read(&obd->u.echo.eo_prep));
 324
 325         RETURN(0);
 326
 327 preprw_cleanup:
 328         /* It is possible that we would rather handle errors by  allow
 329          * any already-set-up pages to complete, rather than tearing them
 330          * all down again.  I believe that this is what the in-kernel
 331          * prep/commit operations do.
 332          */
 333         CERROR("cleaning up %ld pages (%d obdos)\n", (long)(r - res), objcount);
 334         while (r-- > res) {
 335                 kunmap(r->page);
 336                 /* NB if this is an 'object0' page, __free_pages will just
 337                  * lose the extra ref gained above */
 338                 __free_pages(r->page, 0);
 339                 atomic_dec(&obd->u.echo.eo_prep);
 340         }
 341         memset(res, 0, sizeof(*res) * niocount);
 342
 343         return rc;
 344 }
 345
 346 int echo_commitrw(int cmd, struct obd_export *export, struct obdo *oa,
 347                   int objcount, struct obd_ioobj *obj, int niocount,
 348                   struct niobuf_local *res, struct obd_trans_info *oti)
 349 {
 350         struct obd_device *obd;
 351         struct niobuf_local *r = res;
 352         int i, vrc = 0, rc = 0;
 353         ENTRY;
 354
 355         obd = export->exp_obd;
 356         if (obd == NULL)
 357                 RETURN(-EINVAL);
 358
 359         if ((cmd & OBD_BRW_RWMASK) == OBD_BRW_READ) {
 360                 CDEBUG(D_PAGE, "reading %d obdos with %d IOs\n",
 361                        objcount, niocount);
 362         } else {
 363                 CDEBUG(D_PAGE, "writing %d obdos with %d IOs\n",
 364                        objcount, niocount);
 365         }
 366
 367         if (niocount && !r) {
 368                 CERROR("NULL res niobuf with niocount %d\n", niocount);
 369                 RETURN(-EINVAL);
 370         }
 371
 372         LASSERT(oti == NULL || oti->oti_handle == (void *)DESC_PRIV);
 373
 374         for (i = 0; i < objcount; i++, obj++) {
 375                 int verify = obj->ioo_id != 0;
 376                 int j;
 377
 378                 for (j = 0 ; j < obj->ioo_bufcnt ; j++, r++) {
 379                         struct page *page = r->page;
 380                         void *addr;
 381
 382                         if (!page || !(addr = kmap(page)) ||
 383                             !kern_addr_valid(addr)) {
 384
 385                                 CERROR("bad page objid "LPU64":%p, buf %d/%d\n",
 386                                        obj->ioo_id, page, j, obj->ioo_bufcnt);
 387                                 kunmap(page);
 388                                 GOTO(commitrw_cleanup, rc = -EFAULT);
 389                         }
 390
 391                         CDEBUG(D_PAGE, "$$$$ use page %p, addr %p@"LPU64"\n",
 392                                r->page, addr, r->offset);
 393
 394                         if (verify) {
 395                                 vrc = page_debug_check("echo", addr, r->len,
 396                                                        r->offset, obj->ioo_id);
 397                                 /* check all the pages always */
 398                                 if (vrc != 0 && rc == 0)
 399                                         rc = vrc;
 400                         }
 401
 402                         kunmap(page);
 403                         /* NB see comment above regarding object0 pages */
 404                         __free_pages(page, 0);
 405                         atomic_dec(&obd->u.echo.eo_prep);
 406                 }
 407         }
 408         CDEBUG(D_PAGE, "%d pages remain after commit\n",
 409                atomic_read(&obd->u.echo.eo_prep));
 410         RETURN(rc);
 411
 412 commitrw_cleanup:
 413         CERROR("cleaning up %ld pages (%d obdos)\n",
 414                niocount - (long)(r - res) - 1, objcount);
 415         while (++r < res + niocount) {
 416                 struct page *page = r->page;
 417
 418                 /* NB see comment above regarding object0 pages */
 419                 __free_pages(page, 0);
 420                 atomic_dec(&obd->u.echo.eo_prep);
 421         }
 422         return rc;
 423 }
 424
 425 static int echo_setup(struct obd_device *obddev, obd_count len, void *buf)
 426 {
 427         ENTRY;
 428
 429         spin_lock_init(&obddev->u.echo.eo_lock);
 430         obddev->u.echo.eo_lastino = ECHO_INIT_OBJID;
 431
 432         obddev->obd_namespace =
 433                 ldlm_namespace_new("echo-tgt", LDLM_NAMESPACE_SERVER);
 434         if (obddev->obd_namespace == NULL) {
 435                 LBUG();
 436                 RETURN(-ENOMEM);
 437         }
 438
 439         ptlrpc_init_client (LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
 440                             "echo_ldlm_cb_client", &obddev->obd_ldlm_client);
 441         RETURN(0);
 442 }
 443
 444 static int echo_cleanup(struct obd_device *obddev, int flags)
 445 {
 446         ENTRY;
 447
 448         ldlm_namespace_free(obddev->obd_namespace);
 449         CERROR("%d prep/commitrw pages leaked\n",
 450                atomic_read(&obddev->u.echo.eo_prep));
 451
 452         RETURN(0);
 453 }
 454
 455 int echo_attach(struct obd_device *obd, obd_count len, void *data)
 456 {
 457         struct lprocfs_static_vars lvars;
 458         int rc;
 459
 460         lprocfs_init_vars(echo, &lvars);
 461         rc = lprocfs_obd_attach(obd, lvars.obd_vars);
 462         if (rc != 0)
 463                 return rc;
 464         rc = lprocfs_alloc_obd_stats(obd, LPROC_ECHO_LAST);
 465         if (rc != 0)
 466                 return rc;
 467
 468         lprocfs_counter_init(obd->obd_stats, LPROC_ECHO_READ_BYTES,
 469                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
 470         lprocfs_counter_init(obd->obd_stats, LPROC_ECHO_WRITE_BYTES,
 471                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
 472         return rc;
 473 }
 474
 475 int echo_detach(struct obd_device *dev)
 476 {
 477         lprocfs_free_obd_stats(dev);
 478         return lprocfs_obd_detach(dev);
 479 }
 480
 481 static struct obd_ops echo_obd_ops = {
 482         o_owner:       THIS_MODULE,
 483         o_attach:      echo_attach,
 484         o_detach:      echo_detach,
 485         o_connect:     echo_connect,
 486         o_disconnect:  echo_disconnect,
 487         o_create:      echo_create,
 488         o_destroy:     echo_destroy,
 489         o_open:        echo_open,
 490         o_close:       echo_close,
 491         o_getattr:     echo_getattr,
 492         o_setattr:     echo_setattr,
 493         o_preprw:      echo_preprw,
 494         o_commitrw:    echo_commitrw,
 495         o_setup:       echo_setup,
 496         o_cleanup:     echo_cleanup
 497 };
 498
 499 extern int echo_client_init(void);
 500 extern void echo_client_cleanup(void);
 501
 502 static void
 503 echo_object0_pages_fini (void)
 504 {
 505         int     i;
 506
 507         for (i = 0; i < ECHO_OBJECT0_NPAGES; i++)
 508                 if (echo_object0_pages[i] != NULL) {
 509                         __free_pages (echo_object0_pages[i], 0);
 510                         echo_object0_pages[i] = NULL;
 511                 }
 512 }
 513
 514 static int
 515 echo_object0_pages_init (void)
 516 {
 517         struct page *pg;
 518         int          i;
 519
 520         for (i = 0; i < ECHO_OBJECT0_NPAGES; i++) {
 521                 int gfp_mask = (i < ECHO_OBJECT0_NPAGES/2) ?
 522                         GFP_KERNEL : GFP_HIGHUSER;
 523
 524                 pg = alloc_pages (gfp_mask, 0);
 525                 if (pg == NULL) {
 526                         echo_object0_pages_fini ();
 527                         return (-ENOMEM);
 528                 }
 529
 530                 memset (kmap (pg), 0, PAGE_SIZE);
 531                 kunmap (pg);
 532
 533                 echo_object0_pages[i] = pg;
 534         }
 535
 536         return (0);
 537 }
 538
 539 static int __init obdecho_init(void)
 540 {
 541         struct lprocfs_static_vars lvars;
 542         int rc;
 543
 544         printk(KERN_INFO "Lustre Echo OBD driver; info@clusterfs.com\n");
 545
 546         lprocfs_init_vars(echo, &lvars);
 547
 548         rc = echo_object0_pages_init ();
 549         if (rc != 0)
 550                 goto failed_0;
 551
 552         rc = class_register_type(&echo_obd_ops, lvars.module_vars,
 553                                  OBD_ECHO_DEVICENAME);
 554         if (rc != 0)
 555                 goto failed_1;
 556
 557         rc = echo_client_init();
 558         if (rc == 0)
 559                 RETURN (0);
 560
 561         class_unregister_type(OBD_ECHO_DEVICENAME);
 562  failed_1:
 563         echo_object0_pages_fini ();
 564  failed_0:
 565         RETURN(rc);
 566 }
 567
 568 static void /*__exit*/ obdecho_exit(void)
 569 {
 570         echo_client_cleanup();
 571         class_unregister_type(OBD_ECHO_DEVICENAME);
 572         echo_object0_pages_fini ();
 573 }
 574
 575 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 576 MODULE_DESCRIPTION("Lustre Testing Echo OBD driver");
 577 MODULE_LICENSE("GPL");
 578
 579 module_init(obdecho_init);
 580 module_exit(obdecho_exit);