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