lustre/obdecho/echo_client.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  *
   6  *   This file is part of Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #define DEBUG_SUBSYSTEM S_ECHO
  23 #ifdef __KERNEL__
  24 #include <linux/version.h>
  25 #include <linux/module.h>
  26 #include <linux/fs.h>
  27 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
  28 #include <linux/iobuf.h>
  29 #endif
  30 #include <asm/div64.h>
  31 #else
  32 #include <liblustre.h>
  33 #endif
  34
  35 #include <linux/obd.h>
  36 #include <linux/obd_support.h>
  37 #include <linux/obd_class.h>
  38 #include <linux/obd_echo.h>
  39 #include <linux/lustre_debug.h>
  40 #include <linux/lprocfs_status.h>
  41 #include <linux/lustre_lite.h>                  /* for LL_IOC_LOV_SETSTRIPE */
  42
  43 #if 0
  44 static void
  45 echo_printk_object (char *msg, struct ec_object *eco)
  46 {
  47         struct lov_stripe_md *lsm = eco->eco_lsm;
  48         int                   i;
  49
  50         printk (KERN_INFO "%s: object %p: "LPX64", refs %d%s: "LPX64
  51                 "=%u!%u@%d\n", msg, eco, eco->eco_id, eco->eco_refcount,
  52                 eco->eco_deleted ? "(deleted) " : "",
  53                 lsm->lsm_object_id, lsm->lsm_stripe_size,
  54                 lsm->lsm_stripe_count, lsm->lsm_stripe_offset);
  55
  56         for (i = 0; i < lsm->lsm_stripe_count; i++)
  57                 printk (KERN_INFO "   [%2u]"LPX64"\n",
  58                         lsm->lsm_oinfo[i].loi_ost_idx,
  59                         lsm->lsm_oinfo[i].loi_id);
  60 }
  61 #endif
  62
  63 static struct ec_object *
  64 echo_find_object_locked (struct obd_device *obd, obd_id id)
  65 {
  66         struct echo_client_obd *ec = &obd->u.echo_client;
  67         struct ec_object       *eco = NULL;
  68         struct list_head       *el;
  69
  70         list_for_each (el, &ec->ec_objects) {
  71                 eco = list_entry (el, struct ec_object, eco_obj_chain);
  72
  73                 if (eco->eco_id == id)
  74                         return (eco);
  75         }
  76         return (NULL);
  77 }
  78
  79 static int
  80 echo_copyout_lsm (struct lov_stripe_md *lsm, void *ulsm, int ulsm_nob)
  81 {
  82         int nob;
  83
  84         nob = offsetof (struct lov_stripe_md, lsm_oinfo[lsm->lsm_stripe_count]);
  85         if (nob > ulsm_nob)
  86                 return (-EINVAL);
  87
  88         if (copy_to_user (ulsm, lsm, nob))
  89                 return (-EFAULT);
  90
  91         return (0);
  92 }
  93
  94 static int
  95 echo_copyin_lsm (struct obd_device *obd, struct lov_stripe_md *lsm,
  96                  void *ulsm, int ulsm_nob)
  97 {
  98         struct echo_client_obd *ec = &obd->u.echo_client;
  99         int                     nob;
 100
 101         if (ulsm_nob < sizeof (*lsm))
 102                 return (-EINVAL);
 103
 104         if (copy_from_user (lsm, ulsm, sizeof (*lsm)))
 105                 return (-EFAULT);
 106
 107         nob = lsm->lsm_stripe_count * sizeof (lsm->lsm_oinfo[0]);
 108
 109         if (ulsm_nob < nob ||
 110             lsm->lsm_stripe_count > ec->ec_nstripes ||
 111             lsm->lsm_magic != LOV_MAGIC ||
 112             (lsm->lsm_stripe_offset != 0 &&
 113              lsm->lsm_stripe_offset != 0xffffffff &&
 114              lsm->lsm_stripe_offset >= ec->ec_nstripes) ||
 115             (lsm->lsm_stripe_size & (PAGE_SIZE - 1)) != 0 ||
 116             ((__u64)lsm->lsm_stripe_size * lsm->lsm_stripe_count > ~0UL))
 117                 return (-EINVAL);
 118
 119         LASSERT (ec->ec_lsmsize >= sizeof (*lsm) + nob);
 120
 121         if (copy_from_user(lsm->lsm_oinfo,
 122                            ((struct lov_stripe_md *)ulsm)->lsm_oinfo, nob))
 123                 return (-EFAULT);
 124
 125         return (0);
 126 }
 127
 128 static struct ec_object *
 129 echo_allocate_object (struct obd_device *obd)
 130 {
 131         struct echo_client_obd *ec = &obd->u.echo_client;
 132         struct ec_object       *eco;
 133
 134         OBD_ALLOC (eco, sizeof (*eco));
 135         if (eco == NULL)
 136                 return (NULL);
 137
 138         OBD_ALLOC (eco->eco_lsm, ec->ec_lsmsize);
 139         if (eco->eco_lsm == NULL) {
 140                 OBD_FREE (eco, sizeof (*eco));
 141                 return (NULL);
 142         }
 143
 144         eco->eco_device = obd;
 145         eco->eco_deleted = 0;
 146         eco->eco_refcount = 0;
 147         eco->eco_lsm->lsm_magic = LOV_MAGIC;
 148         /* leave stripe count 0 by default */
 149
 150         return (eco);
 151 }
 152
 153 static void
 154 echo_free_object (struct ec_object *eco)
 155 {
 156         struct obd_device      *obd = eco->eco_device;
 157         struct echo_client_obd *ec = &obd->u.echo_client;
 158
 159         LASSERT (eco->eco_refcount == 0);
 160         OBD_FREE (eco->eco_lsm, ec->ec_lsmsize);
 161         OBD_FREE (eco, sizeof (*eco));
 162 }
 163
 164 static int
 165 echo_create_object (struct obd_device *obd, int on_target, struct obdo *oa,
 166                     void *ulsm, int ulsm_nob)
 167 {
 168         struct echo_client_obd *ec = &obd->u.echo_client;
 169         struct ec_object       *eco2;
 170         struct ec_object       *eco;
 171         struct lov_stripe_md   *lsm;
 172         int                     rc;
 173         int                     i;
 174
 175         if ((oa->o_valid & OBD_MD_FLID) == 0 && /* no obj id */
 176             (on_target ||                       /* set_stripe */
 177              ec->ec_nstripes != 0)) {           /* LOV */
 178                 CERROR ("No valid oid\n");
 179                 return (-EINVAL);
 180         }
 181
 182         eco = echo_allocate_object (obd);
 183         if (eco == NULL)
 184                 return (-ENOMEM);
 185
 186         lsm = eco->eco_lsm;
 187
 188         if (ulsm != NULL) {
 189                 rc = echo_copyin_lsm (obd, lsm, ulsm, ulsm_nob);
 190                 if (rc != 0)
 191                         goto failed;
 192         }
 193
 194         /* setup object ID here for !on_target and LOV hint */
 195         if ((oa->o_valid & OBD_MD_FLID) != 0)
 196                 eco->eco_id = lsm->lsm_object_id = oa->o_id;
 197
 198         /* defaults -> actual values */
 199         if (lsm->lsm_stripe_offset == 0xffffffff)
 200                 lsm->lsm_stripe_offset = 0;
 201
 202         if (lsm->lsm_stripe_count == 0)
 203                 lsm->lsm_stripe_count = ec->ec_nstripes;
 204
 205         if (lsm->lsm_stripe_size == 0)
 206                 lsm->lsm_stripe_size = PAGE_SIZE;
 207
 208         /* setup stripes: indices + default ids if required */
 209         for (i = 0; i < lsm->lsm_stripe_count; i++) {
 210                 if (lsm->lsm_oinfo[i].loi_id == 0)
 211                         lsm->lsm_oinfo[i].loi_id = lsm->lsm_object_id;
 212
 213                 lsm->lsm_oinfo[i].loi_ost_idx =
 214                         (lsm->lsm_stripe_offset + i) % ec->ec_nstripes;
 215         }
 216
 217         if (on_target) {
 218                 rc = obd_create (&ec->ec_conn, oa, &lsm, NULL);
 219                 if (rc != 0)
 220                         goto failed;
 221
 222                 /* See what object ID we were given */
 223                 LASSERT ((oa->o_valid & OBD_MD_FLID) != 0);
 224                 eco->eco_id = lsm->lsm_object_id = oa->o_id;
 225         }
 226
 227         spin_lock (&ec->ec_lock);
 228
 229         eco2 = echo_find_object_locked (obd, oa->o_id);
 230         if (eco2 != NULL) {                     /* conflict */
 231                 spin_unlock (&ec->ec_lock);
 232
 233                 CERROR ("Can't create object id "LPX64": id already exists%s\n",
 234                         oa->o_id, on_target ? " (undoing create)" : "");
 235
 236                 if (on_target)
 237                         obd_destroy (&ec->ec_conn, oa, lsm, NULL);
 238
 239                 rc = -EEXIST;
 240                 goto failed;
 241         }
 242
 243         list_add (&eco->eco_obj_chain, &ec->ec_objects);
 244         spin_unlock (&ec->ec_lock);
 245         CDEBUG (D_INFO,
 246                 "created %p: "LPX64"=%u#%u&%d refs %d del %d\n",
 247                 eco, eco->eco_id,
 248                 eco->eco_lsm->lsm_stripe_size,
 249                 eco->eco_lsm->lsm_stripe_count,
 250                 eco->eco_lsm->lsm_stripe_offset,
 251                 eco->eco_refcount, eco->eco_deleted);
 252         return (0);
 253
 254  failed:
 255         echo_free_object (eco);
 256         return (rc);
 257 }
 258
 259 static int
 260 echo_get_object (struct ec_object **ecop, struct obd_device *obd,
 261                  struct obdo *oa)
 262 {
 263         struct echo_client_obd *ec = &obd->u.echo_client;
 264         struct ec_object       *eco;
 265         struct ec_object       *eco2;
 266         int                     rc;
 267
 268         if ((oa->o_valid & OBD_MD_FLID) == 0)
 269         {
 270                 CERROR ("No valid oid\n");
 271                 return (-EINVAL);
 272         }
 273
 274         spin_lock (&ec->ec_lock);
 275         eco = echo_find_object_locked (obd, oa->o_id);
 276         if (eco != NULL) {
 277                 if (eco->eco_deleted)           /* being deleted */
 278                         return (-EAGAIN);       /* (see comment in cleanup) */
 279
 280                 eco->eco_refcount++;
 281                 spin_unlock (&ec->ec_lock);
 282                 *ecop = eco;
 283                 CDEBUG (D_INFO,
 284                         "found %p: "LPX64"=%u#%u&%d refs %d del %d\n",
 285                         eco, eco->eco_id,
 286                         eco->eco_lsm->lsm_stripe_size,
 287                         eco->eco_lsm->lsm_stripe_count,
 288                         eco->eco_lsm->lsm_stripe_offset,
 289                         eco->eco_refcount, eco->eco_deleted);
 290                 return (0);
 291         }
 292         spin_unlock (&ec->ec_lock);
 293
 294         if (ec->ec_nstripes != 0)               /* striping required */
 295                 return (-ENOENT);
 296
 297         eco = echo_allocate_object (obd);
 298         if (eco == NULL)
 299                 return (-ENOMEM);
 300
 301         eco->eco_id = eco->eco_lsm->lsm_object_id = oa->o_id;
 302
 303         spin_lock (&ec->ec_lock);
 304
 305         eco2 = echo_find_object_locked (obd, oa->o_id);
 306         if (eco2 == NULL) {                     /* didn't race */
 307                 list_add (&eco->eco_obj_chain, &ec->ec_objects);
 308                 spin_unlock (&ec->ec_lock);
 309                 eco->eco_refcount = 1;
 310                 *ecop = eco;
 311                 CDEBUG (D_INFO,
 312                         "created %p: "LPX64"=%u#%u&%d refs %d del %d\n",
 313                         eco, eco->eco_id,
 314                         eco->eco_lsm->lsm_stripe_size,
 315                         eco->eco_lsm->lsm_stripe_count,
 316                         eco->eco_lsm->lsm_stripe_offset,
 317                         eco->eco_refcount, eco->eco_deleted);
 318                 return (0);
 319         }
 320
 321         if (eco2->eco_deleted)
 322                 rc = -EAGAIN;                   /* lose race */
 323         else {
 324                 eco2->eco_refcount++;           /* take existing */
 325                 *ecop = eco2;
 326                 rc = 0;
 327                 LASSERT (eco2->eco_id == eco2->eco_lsm->lsm_object_id);
 328                 CDEBUG (D_INFO,
 329                         "found(2) %p: "LPX64"=%u#%u&%d refs %d del %d\n",
 330                         eco2, eco2->eco_id,
 331                         eco2->eco_lsm->lsm_stripe_size,
 332                         eco2->eco_lsm->lsm_stripe_count,
 333                         eco2->eco_lsm->lsm_stripe_offset,
 334                         eco2->eco_refcount, eco2->eco_deleted);
 335         }
 336
 337         spin_unlock (&ec->ec_lock);
 338
 339         echo_free_object (eco);
 340         return (rc);
 341 }
 342
 343 static void
 344 echo_put_object (struct ec_object *eco)
 345 {
 346         struct obd_device      *obd = eco->eco_device;
 347         struct echo_client_obd *ec = &obd->u.echo_client;
 348
 349         /* Release caller's ref on the object.
 350          * delete => mark for deletion when last ref goes
 351          */
 352
 353         spin_lock (&ec->ec_lock);
 354
 355         eco->eco_refcount--;
 356         LASSERT (eco->eco_refcount >= 0);
 357
 358         CDEBUG(D_INFO, "put %p: "LPX64"=%u#%u&%d refs %d del %d\n",
 359                eco, eco->eco_id,
 360                eco->eco_lsm->lsm_stripe_size,
 361                eco->eco_lsm->lsm_stripe_count,
 362                eco->eco_lsm->lsm_stripe_offset,
 363                eco->eco_refcount, eco->eco_deleted);
 364
 365         if (eco->eco_refcount != 0 || !eco->eco_deleted) {
 366                 spin_unlock (&ec->ec_lock);
 367                 return;
 368         }
 369
 370         spin_unlock (&ec->ec_lock);
 371
 372         /* NB leave obj in the object list.  We must prevent anyone from
 373          * attempting to enqueue on this object number until we can be
 374          * sure there will be no more lock callbacks.
 375          */
 376         obd_cancel_unused(&ec->ec_conn, eco->eco_lsm, 0, NULL);
 377
 378         /* now we can let it go */
 379         spin_lock (&ec->ec_lock);
 380         list_del (&eco->eco_obj_chain);
 381         spin_unlock (&ec->ec_lock);
 382
 383         LASSERT (eco->eco_refcount == 0);
 384
 385         echo_free_object (eco);
 386 }
 387
 388 static void
 389 echo_get_stripe_off_id (struct lov_stripe_md *lsm, obd_off *offp, obd_id *idp)
 390 {
 391         unsigned long stripe_count;
 392         unsigned long stripe_size;
 393         unsigned long width;
 394         unsigned long woffset;
 395         int           stripe_index;
 396         obd_off       offset;
 397
 398         if (lsm->lsm_stripe_count <= 1)
 399                 return;
 400
 401         offset       = *offp;
 402         stripe_size  = lsm->lsm_stripe_size;
 403         stripe_count = lsm->lsm_stripe_count;
 404
 405         /* width = # bytes in all stripes */
 406         width = stripe_size * stripe_count;
 407
 408         /* woffset = offset within a width; offset = whole number of widths */
 409         woffset = do_div (offset, width);
 410
 411         stripe_index = woffset / stripe_size;
 412
 413         *idp = lsm->lsm_oinfo[stripe_index].loi_id;
 414         *offp = offset * stripe_size + woffset % stripe_size;
 415 }
 416
 417 static int
 418 echo_client_kbrw (struct obd_device *obd, int rw,
 419                   struct obdo *oa, struct lov_stripe_md *lsm,
 420                   obd_off offset, obd_size count)
 421 {
 422         struct echo_client_obd *ec = &obd->u.echo_client;
 423         obd_count               npages;
 424         struct brw_page        *pga;
 425         struct brw_page        *pgp;
 426         obd_off                 off;
 427         int                     i;
 428         int                     rc;
 429         int                     verify;
 430         int                     gfp_mask;
 431
 432         /* oa_id  == 0    => speed test (no verification) else...
 433          * oa & 1         => use HIGHMEM
 434          */
 435         verify = (oa->o_id != 0);
 436         gfp_mask = ((oa->o_id & 1) == 0) ? GFP_KERNEL : GFP_HIGHUSER;
 437
 438         LASSERT(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ);
 439
 440         if (count <= 0 ||
 441             (count & (PAGE_SIZE - 1)) != 0 ||
 442             (lsm != NULL &&
 443              lsm->lsm_object_id != oa->o_id))
 444                 return (-EINVAL);
 445
 446         /* XXX think again with misaligned I/O */
 447         npages = count >> PAGE_SHIFT;
 448
 449         OBD_ALLOC(pga, npages * sizeof(*pga));
 450         if (pga == NULL)
 451                 return (-ENOMEM);
 452
 453         for (i = 0, pgp = pga, off = offset;
 454              i < npages;
 455              i++, pgp++, off += PAGE_SIZE) {
 456
 457                 LASSERT (pgp->pg == NULL);      /* for cleanup */
 458
 459                 rc = -ENOMEM;
 460                 pgp->pg = alloc_pages (gfp_mask, 0);
 461                 if (pgp->pg == NULL)
 462                         goto out;
 463
 464                 pgp->count = PAGE_SIZE;
 465                 pgp->off = off;
 466                 pgp->flag = 0;
 467
 468                 if (verify) {
 469                         void *addr = kmap(pgp->pg);
 470                         obd_off      stripe_off = off;
 471                         obd_id       stripe_id = oa->o_id;
 472
 473                         if (rw == OBD_BRW_WRITE) {
 474                                 echo_get_stripe_off_id(lsm, &stripe_off,
 475                                                        &stripe_id);
 476                                 page_debug_setup(addr, pgp->count,
 477                                                  stripe_off, stripe_id);
 478                         } else {
 479                                 page_debug_setup(addr, pgp->count,
 480                                                  0xdeadbeef00c0ffee,
 481                                                  0xdeadbeef00c0ffee);
 482                         }
 483                         kunmap(pgp->pg);
 484                 }
 485         }
 486
 487         rc = obd_brw(rw, &ec->ec_conn, lsm, npages, pga, NULL);
 488
 489  out:
 490         if (rc != 0)
 491                 verify = 0;
 492
 493         for (i = 0, pgp = pga; i < npages; i++, pgp++) {
 494                 if (pgp->pg == NULL)
 495                         continue;
 496
 497                 if (verify) {
 498                         void    *addr = kmap(pgp->pg);
 499                         obd_off  stripe_off = pgp->off;
 500                         obd_id   stripe_id  = oa->o_id;
 501                         int      vrc;
 502
 503                         echo_get_stripe_off_id (lsm, &stripe_off, &stripe_id);
 504                         vrc = page_debug_check("test_brw", addr, pgp->count,
 505                                                stripe_off, stripe_id);
 506                         if (vrc != 0 && rc == 0)
 507                                 rc = vrc;
 508
 509                         kunmap(pgp->pg);
 510                 }
 511                 __free_pages(pgp->pg, 0);
 512         }
 513         OBD_FREE(pga, npages * sizeof(*pga));
 514         return (rc);
 515 }
 516
 517 #ifdef __KERNEL__
 518 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 519 static int echo_client_ubrw(struct obd_device *obd, int rw,
 520                             struct obdo *oa, struct lov_stripe_md *lsm,
 521                             obd_off offset, obd_size count, char *buffer)
 522 {
 523         struct echo_client_obd *ec = &obd->u.echo_client;
 524         obd_count               npages;
 525         struct brw_page        *pga;
 526         struct brw_page        *pgp;
 527         obd_off                 off;
 528         struct kiobuf          *kiobuf;
 529         int                     i;
 530         int                     rc;
 531
 532         LASSERT (rw == OBD_BRW_WRITE ||
 533                  rw == OBD_BRW_READ);
 534
 535         /* NB: for now, only whole pages, page aligned */
 536
 537         if (count <= 0 ||
 538             ((long)buffer & (PAGE_SIZE - 1)) != 0 ||
 539             (count & (PAGE_SIZE - 1)) != 0 ||
 540             (lsm != NULL && lsm->lsm_object_id != oa->o_id))
 541                 return (-EINVAL);
 542
 543         /* XXX think again with misaligned I/O */
 544         npages = count >> PAGE_SHIFT;
 545
 546         OBD_ALLOC(pga, npages * sizeof(*pga));
 547         if (pga == NULL)
 548                 return (-ENOMEM);
 549
 550         rc = alloc_kiovec (1, &kiobuf);
 551         if (rc != 0)
 552                 goto out_1;
 553
 554         rc = map_user_kiobuf ((rw == OBD_BRW_READ) ? READ : WRITE,
 555                               kiobuf, (unsigned long)buffer, count);
 556         if (rc != 0)
 557                 goto out_2;
 558
 559         LASSERT (kiobuf->offset == 0);
 560         LASSERT (kiobuf->nr_pages == npages);
 561
 562         for (i = 0, off = offset, pgp = pga;
 563              i < npages;
 564              i++, off += PAGE_SIZE, pgp++) {
 565                 pgp->off = off;
 566                 pgp->pg = kiobuf->maplist[i];
 567                 pgp->count = PAGE_SIZE;
 568                 pgp->flag = 0;
 569         }
 570
 571         rc = obd_brw(rw, &ec->ec_conn, lsm, npages, pga, NULL);
 572
 573         //        if (rw == OBD_BRW_READ)
 574         //                mark_dirty_kiobuf (kiobuf, count);
 575
 576         unmap_kiobuf (kiobuf);
 577  out_2:
 578         free_kiovec (1, &kiobuf);
 579  out_1:
 580         OBD_FREE(pga, npages * sizeof(*pga));
 581         return (rc);
 582 }
 583 #else
 584 static int echo_client_ubrw(struct obd_device *obd, int rw,
 585                             struct obdo *oa, struct lov_stripe_md *lsm,
 586                             obd_off offset, obd_size count, char *buffer)
 587 {
 588         LBUG();
 589         return 0;
 590 }
 591 #endif
 592 #endif
 593
 594 static int
 595 echo_open (struct obd_export *exp, struct obdo *oa)
 596 {
 597         struct obd_device      *obd = exp->exp_obd;
 598         struct echo_client_obd *ec = &obd->u.echo_client;
 599         struct lustre_handle   *ufh = obdo_handle (oa);
 600         struct ec_open_object  *ecoo;
 601         struct ec_object       *eco;
 602         int                     rc;
 603
 604         rc = echo_get_object (&eco, obd, oa);
 605         if (rc != 0)
 606                 return rc;
 607
 608         rc = -ENOMEM;
 609         OBD_ALLOC (ecoo, sizeof (*ecoo));
 610         if (ecoo == NULL)
 611                 goto failed_0;
 612
 613         rc = obd_open(&ec->ec_conn, oa, eco->eco_lsm, NULL, &ecoo->ecoo_och);
 614         if (rc != 0)
 615                 goto failed_1;
 616
 617         memcpy (&ecoo->ecoo_oa, oa, sizeof (*oa));
 618         ecoo->ecoo_object = eco;
 619         /* ecoo takes ref from echo_get_object() above */
 620
 621         spin_lock (&ec->ec_lock);
 622
 623         list_add (&ecoo->ecoo_exp_chain, &exp->exp_ec_data.eced_open_head);
 624         ufh->cookie = ecoo->ecoo_cookie = ec->ec_unique++;
 625         spin_unlock (&ec->ec_lock);
 626         return 0;
 627
 628  failed_1:
 629         OBD_FREE (ecoo, sizeof (*ecoo));
 630  failed_0:
 631         echo_put_object (eco);
 632         return (rc);
 633 }
 634
 635 static int
 636 echo_close (struct obd_export *exp, struct obdo *oa)
 637 {
 638         struct obd_device      *obd = exp->exp_obd;
 639         struct echo_client_obd *ec = &obd->u.echo_client;
 640         struct lustre_handle   *ufh = obdo_handle (oa);
 641         struct ec_open_object  *ecoo = NULL;
 642         int                     found = 0;
 643         struct list_head       *el;
 644         int                     rc;
 645
 646         if ((oa->o_valid & OBD_MD_FLHANDLE) == 0)
 647                 return -EINVAL;
 648
 649         spin_lock (&ec->ec_lock);
 650
 651         list_for_each (el, &exp->exp_ec_data.eced_open_head) {
 652                 ecoo = list_entry (el, struct ec_open_object, ecoo_exp_chain);
 653                 found = (ecoo->ecoo_cookie == ufh->cookie);
 654                 if (found) {
 655                         list_del (&ecoo->ecoo_exp_chain);
 656                         break;
 657                 }
 658         }
 659
 660         spin_unlock (&ec->ec_lock);
 661
 662         memcpy(&ecoo->ecoo_oa.o_inline, &ecoo->ecoo_och, FD_OSTDATA_SIZE);
 663         ecoo->ecoo_oa.o_valid |= OBD_MD_FLHANDLE;
 664
 665         rc = obd_close (&ec->ec_conn, &ecoo->ecoo_oa,
 666                         ecoo->ecoo_object->eco_lsm, NULL);
 667
 668         echo_put_object (ecoo->ecoo_object);
 669         OBD_FREE (ecoo, sizeof (*ecoo));
 670
 671         return (rc);
 672 }
 673
 674 static int
 675 echo_ldlm_callback (struct ldlm_lock *lock, struct ldlm_lock_desc *new,
 676                     void *data, int flag)
 677 {
 678         struct ec_object       *eco = (struct ec_object *)data;
 679         struct echo_client_obd *ec = &(eco->eco_device->u.echo_client);
 680         struct lustre_handle    lockh;
 681         struct list_head       *el;
 682         int                     found = 0;
 683         int                     rc;
 684
 685         ldlm_lock2handle (lock, &lockh);
 686
 687         /* #ifdef this out if we're not feeling paranoid */
 688         spin_lock (&ec->ec_lock);
 689         list_for_each (el, &ec->ec_objects) {
 690                 found = (eco == list_entry(el, struct ec_object,
 691                                            eco_obj_chain));
 692                 if (found)
 693                         break;
 694         }
 695         spin_unlock (&ec->ec_lock);
 696         LASSERT (found);
 697
 698         switch (flag) {
 699         case LDLM_CB_BLOCKING:
 700                 CDEBUG(D_INFO, "blocking callback on "LPX64", handle "LPX64"\n",
 701                        eco->eco_id, lockh.cookie);
 702                 rc = ldlm_cli_cancel (&lockh);
 703                 if (rc != ELDLM_OK)
 704                         CERROR ("ldlm_cli_cancel failed: %d\n", rc);
 705                 break;
 706
 707         case LDLM_CB_CANCELING:
 708                 CDEBUG(D_INFO, "cancel callback on "LPX64", handle "LPX64"\n",
 709                        eco->eco_id, lockh.cookie);
 710                 break;
 711
 712         default:
 713                 LBUG ();
 714         }
 715
 716         return (0);
 717 }
 718
 719 static int
 720 echo_enqueue (struct obd_export *exp, struct obdo *oa,
 721               int mode, obd_off offset, obd_size nob)
 722 {
 723         struct obd_device      *obd = exp->exp_obd;
 724         struct echo_client_obd *ec = &obd->u.echo_client;
 725         struct lustre_handle   *ulh = obdo_handle (oa);
 726         struct ec_object       *eco;
 727         struct ec_lock         *ecl;
 728         int                     flags;
 729         int                     rc;
 730
 731         if (!(mode == LCK_PR || mode == LCK_PW))
 732                 return -EINVAL;
 733
 734         if ((offset & (PAGE_SIZE - 1)) != 0 ||
 735             (nob & (PAGE_SIZE - 1)) != 0)
 736                 return -EINVAL;
 737
 738         rc = echo_get_object (&eco, obd, oa);
 739         if (rc != 0)
 740                 return rc;
 741
 742         rc = -ENOMEM;
 743         OBD_ALLOC (ecl, sizeof (*ecl));
 744         if (ecl == NULL)
 745                 goto failed_0;
 746
 747         ecl->ecl_mode = mode;
 748         ecl->ecl_object = eco;
 749         ecl->ecl_extent.start = offset;
 750         ecl->ecl_extent.end = (nob == 0) ? ((obd_off) -1) : (offset + nob - 1);
 751
 752         flags = 0;
 753         rc = obd_enqueue(&ec->ec_conn, eco->eco_lsm, NULL, LDLM_EXTENT,
 754                          &ecl->ecl_extent,sizeof(ecl->ecl_extent), mode,
 755                          &flags, echo_ldlm_callback, eco,
 756                          &ecl->ecl_lock_handle);
 757         if (rc != 0)
 758                 goto failed_1;
 759
 760         CDEBUG(D_INFO, "enqueue handle "LPX64"\n", ecl->ecl_lock_handle.cookie);
 761
 762         /* NB ecl takes object ref from echo_get_object() above */
 763         spin_lock(&ec->ec_lock);
 764
 765         list_add(&ecl->ecl_exp_chain, &exp->exp_ec_data.eced_locks);
 766         ulh->cookie = ecl->ecl_cookie = ec->ec_unique++;
 767
 768         spin_unlock(&ec->ec_lock);
 769
 770         oa->o_valid |= OBD_MD_FLHANDLE;
 771         return 0;
 772
 773  failed_1:
 774         OBD_FREE (ecl, sizeof (*ecl));
 775  failed_0:
 776         echo_put_object (eco);
 777         return (rc);
 778 }
 779
 780 static int
 781 echo_cancel (struct obd_export *exp, struct obdo *oa)
 782 {
 783         struct obd_device      *obd = exp->exp_obd;
 784         struct echo_client_obd *ec = &obd->u.echo_client;
 785         struct lustre_handle   *ulh = obdo_handle (oa);
 786         struct ec_lock         *ecl = NULL;
 787         int                     found = 0;
 788         struct list_head       *el;
 789         int                     rc;
 790
 791         if ((oa->o_valid & OBD_MD_FLHANDLE) == 0)
 792                 return -EINVAL;
 793
 794         spin_lock (&ec->ec_lock);
 795
 796         list_for_each (el, &exp->exp_ec_data.eced_locks) {
 797                 ecl = list_entry (el, struct ec_lock, ecl_exp_chain);
 798                 found = (ecl->ecl_cookie == ulh->cookie);
 799                 if (found) {
 800                         list_del (&ecl->ecl_exp_chain);
 801                         break;
 802                 }
 803         }
 804
 805         spin_unlock (&ec->ec_lock);
 806
 807         if (!found)
 808                 return (-ENOENT);
 809
 810         rc = obd_cancel(&ec->ec_conn, ecl->ecl_object->eco_lsm, ecl->ecl_mode,
 811                         &ecl->ecl_lock_handle);
 812
 813         echo_put_object (ecl->ecl_object);
 814         OBD_FREE (ecl, sizeof (*ecl));
 815
 816         return rc;
 817 }
 818
 819 static int echo_iocontrol(unsigned int cmd, struct lustre_handle *obdconn,
 820                           int len, void *karg, void *uarg)
 821 {
 822         struct obd_export      *exp = class_conn2export (obdconn);
 823         struct obd_device      *obd;
 824         struct echo_client_obd *ec;
 825         struct ec_object       *eco;
 826         struct obd_ioctl_data  *data = karg;
 827         int                     rw = OBD_BRW_READ;
 828         int                     rc = 0;
 829         ENTRY;
 830
 831         if (exp == NULL) {
 832                 CERROR("ioctl: No device\n");
 833                 GOTO(out, rc = -EINVAL);
 834         }
 835
 836         obd = exp->exp_obd;
 837         ec = &obd->u.echo_client;
 838
 839         switch (cmd) {
 840         case OBD_IOC_CREATE:                    /* may create echo object */
 841                 if (!capable (CAP_SYS_ADMIN))
 842                         GOTO (out, rc = -EPERM);
 843
 844                 rc = echo_create_object (obd, 1, &data->ioc_obdo1,
 845                                          data->ioc_pbuf1, data->ioc_plen1);
 846                 GOTO(out, rc);
 847
 848         case OBD_IOC_DESTROY:
 849                 if (!capable (CAP_SYS_ADMIN))
 850                         GOTO (out, rc = -EPERM);
 851
 852                 rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
 853                 if (rc == 0) {
 854                         rc = obd_destroy(&ec->ec_conn, &data->ioc_obdo1,
 855                                          eco->eco_lsm, NULL);
 856                         if (rc == 0)
 857                                 eco->eco_deleted = 1;
 858                         echo_put_object(eco);
 859                 }
 860                 GOTO(out, rc);
 861
 862         case OBD_IOC_GETATTR:
 863                 rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
 864                 if (rc == 0) {
 865                         rc = obd_getattr(&ec->ec_conn, &data->ioc_obdo1,
 866                                          eco->eco_lsm);
 867                         echo_put_object(eco);
 868                 }
 869                 GOTO(out, rc);
 870
 871         case OBD_IOC_SETATTR:
 872                 if (!capable (CAP_SYS_ADMIN))
 873                         GOTO (out, rc = -EPERM);
 874
 875                 rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
 876                 if (rc == 0) {
 877                         rc = obd_setattr(&ec->ec_conn, &data->ioc_obdo1,
 878                                          eco->eco_lsm, NULL);
 879                         echo_put_object(eco);
 880                 }
 881                 GOTO(out, rc);
 882
 883         case OBD_IOC_OPEN:
 884                 rc = echo_open (exp, &data->ioc_obdo1);
 885                 GOTO(out, rc);
 886
 887         case OBD_IOC_CLOSE:
 888                 rc = echo_close (exp, &data->ioc_obdo1);
 889                 GOTO(out, rc);
 890
 891         case OBD_IOC_BRW_WRITE:
 892                 if (!capable (CAP_SYS_ADMIN))
 893                         GOTO (out, rc = -EPERM);
 894
 895                 rw = OBD_BRW_WRITE;
 896                 /* fall through */
 897         case OBD_IOC_BRW_READ:
 898                 rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
 899                 if (rc == 0) {
 900                         if (data->ioc_pbuf2 == NULL) // NULL user data pointer
 901                                 rc = echo_client_kbrw(obd, rw, &data->ioc_obdo1,
 902                                                       eco->eco_lsm,
 903                                                       data->ioc_offset,
 904                                                       data->ioc_count);
 905                         else
 906 #ifdef __KERNEL__
 907                                 rc = echo_client_ubrw(obd, rw, &data->ioc_obdo1,
 908                                                       eco->eco_lsm,
 909                                                       data->ioc_offset,
 910                                                       data->ioc_count,
 911                                                       data->ioc_pbuf2);
 912 #endif
 913                         echo_put_object(eco);
 914                 }
 915                 GOTO(out, rc);
 916
 917         case ECHO_IOC_GET_STRIPE:
 918                 rc = echo_get_object(&eco, obd, &data->ioc_obdo1);
 919                 if (rc == 0) {
 920                         rc = echo_copyout_lsm(eco->eco_lsm, data->ioc_pbuf1,
 921                                               data->ioc_plen1);
 922                         echo_put_object(eco);
 923                 }
 924                 GOTO(out, rc);
 925
 926         case ECHO_IOC_SET_STRIPE:
 927                 if (!capable (CAP_SYS_ADMIN))
 928                         GOTO (out, rc = -EPERM);
 929
 930                 if (data->ioc_pbuf1 == NULL) {  /* unset */
 931                         rc = echo_get_object(&eco, obd, &data->ioc_obdo1);
 932                         if (rc == 0) {
 933                                 eco->eco_deleted = 1;
 934                                 echo_put_object(eco);
 935                         }
 936                 } else {
 937                         rc = echo_create_object(obd, 0, &data->ioc_obdo1,
 938                                                 data->ioc_pbuf1,
 939                                                 data->ioc_plen1);
 940                 }
 941                 GOTO (out, rc);
 942
 943         case ECHO_IOC_ENQUEUE:
 944                 if (!capable (CAP_SYS_ADMIN))
 945                         GOTO (out, rc = -EPERM);
 946
 947                 rc = echo_enqueue (exp, &data->ioc_obdo1,
 948                                    data->ioc_conn1, /* lock mode */
 949                                    data->ioc_offset, data->ioc_count);/*extent*/
 950                 GOTO (out, rc);
 951
 952         case ECHO_IOC_CANCEL:
 953                 rc = echo_cancel (exp, &data->ioc_obdo1);
 954                 GOTO (out, rc);
 955
 956         default:
 957                 CERROR ("echo_ioctl(): unrecognised ioctl %#x\n", cmd);
 958                 GOTO (out, rc = -ENOTTY);
 959         }
 960
 961         EXIT;
 962  out:
 963         class_export_put(exp);
 964         return rc;
 965 }
 966
 967 static int echo_setup(struct obd_device *obddev, obd_count len, void *buf)
 968 {
 969         struct obd_ioctl_data* data = buf;
 970         struct echo_client_obd *ec = &obddev->u.echo_client;
 971         struct obd_device *tgt;
 972         struct obd_uuid uuid;
 973         struct lov_stripe_md *lsm = NULL;
 974         struct obd_uuid echo_uuid = { "ECHO_UUID" };
 975         int rc;
 976         ENTRY;
 977
 978         if (data->ioc_inllen1 < 1) {
 979                 CERROR("requires a TARGET OBD UUID\n");
 980                 RETURN(-EINVAL);
 981         }
 982         if (data->ioc_inllen1 > 37) {
 983                 CERROR("OBD UUID must be less than 38 characters\n");
 984                 RETURN(-EINVAL);
 985         }
 986
 987         obd_str2uuid(&uuid, data->ioc_inlbuf1);
 988         tgt = class_uuid2obd(&uuid);
 989         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
 990                 CERROR("device not attached or not set up (%d)\n",
 991                        data->ioc_dev);
 992                 RETURN(rc = -EINVAL);
 993         }
 994
 995         spin_lock_init (&ec->ec_lock);
 996         INIT_LIST_HEAD (&ec->ec_objects);
 997         ec->ec_unique = 0;
 998
 999         rc = obd_connect(&ec->ec_conn, tgt, &echo_uuid);
1000         if (rc) {
1001                 CERROR("fail to connect to device %d\n", data->ioc_dev);
1002                 return (rc);
1003         }
1004
1005         ec->ec_lsmsize = obd_alloc_memmd (&ec->ec_conn, &lsm);
1006         if (ec->ec_lsmsize < 0) {
1007                 CERROR ("Can't get # stripes: %d\n", rc);
1008                 obd_disconnect (&ec->ec_conn, 0);
1009                 rc = ec->ec_lsmsize;
1010         } else {
1011                 ec->ec_nstripes = lsm->lsm_stripe_count;
1012                 obd_free_memmd (&ec->ec_conn, &lsm);
1013         }
1014
1015         RETURN(rc);
1016 }
1017
1018 static int echo_cleanup(struct obd_device * obddev, int force, int failover)
1019 {
1020         struct list_head       *el;
1021         struct ec_object       *eco;
1022         struct echo_client_obd *ec = &obddev->u.echo_client;
1023         int rc;
1024         ENTRY;
1025
1026         if (!list_empty(&obddev->obd_exports)) {
1027                 CERROR("still has clients!\n");
1028                 RETURN(-EBUSY);
1029         }
1030
1031         /* XXX assuming sole access */
1032         while (!list_empty (&ec->ec_objects)) {
1033                 el = ec->ec_objects.next;
1034                 eco = list_entry (el, struct ec_object, eco_obj_chain);
1035
1036                 LASSERT (eco->eco_refcount == 0);
1037                 eco->eco_refcount = 1;
1038                 eco->eco_deleted = 1;
1039                 echo_put_object (eco);
1040         }
1041
1042         rc = obd_disconnect (&ec->ec_conn, 0);
1043         if (rc != 0)
1044                 CERROR("fail to disconnect device: %d\n", rc);
1045
1046         RETURN (rc);
1047 }
1048
1049 static int echo_connect(struct lustre_handle *conn, struct obd_device *src,
1050                         struct obd_uuid *cluuid)
1051 {
1052         struct obd_export *exp;
1053         int                rc;
1054
1055         rc = class_connect(conn, src, cluuid);
1056         if (rc == 0) {
1057                 exp = class_conn2export (conn);
1058                 INIT_LIST_HEAD(&exp->exp_ec_data.eced_open_head);
1059                 INIT_LIST_HEAD(&exp->exp_ec_data.eced_locks);
1060                 class_export_put(exp);
1061         }
1062
1063         RETURN (rc);
1064 }
1065
1066 static int echo_disconnect(struct lustre_handle *conn, int failover)
1067 {
1068         struct obd_export      *exp = class_conn2export (conn);
1069         struct obd_device      *obd;
1070         struct echo_client_obd *ec;
1071         struct ec_open_object  *ecoo;
1072         struct ec_lock         *ecl;
1073         int                     rc;
1074
1075         if (exp == NULL)
1076                 GOTO(out, rc = -EINVAL);
1077
1078         obd = exp->exp_obd;
1079         ec = &obd->u.echo_client;
1080
1081         /* no more contention on export's lock list */
1082         while (!list_empty (&exp->exp_ec_data.eced_locks)) {
1083                 ecl = list_entry (exp->exp_ec_data.eced_locks.next,
1084                                   struct ec_lock, ecl_exp_chain);
1085                 list_del (&ecl->ecl_exp_chain);
1086
1087                 rc = obd_cancel (&ec->ec_conn, ecl->ecl_object->eco_lsm,
1088                                  ecl->ecl_mode, &ecl->ecl_lock_handle);
1089
1090                 CDEBUG (D_INFO, "Cancel lock on object "LPX64" on disconnect (%d)\n",
1091                         ecl->ecl_object->eco_id, rc);
1092
1093                 echo_put_object (ecl->ecl_object);
1094                 OBD_FREE (ecl, sizeof (*ecl));
1095         }
1096
1097         /* no more contention on export's open handle list  */
1098         while (!list_empty (&exp->exp_ec_data.eced_open_head)) {
1099                 ecoo = list_entry (exp->exp_ec_data.eced_open_head.next,
1100                                    struct ec_open_object, ecoo_exp_chain);
1101                 list_del (&ecoo->ecoo_exp_chain);
1102
1103                 memcpy (&ecoo->ecoo_oa.o_inline, &ecoo->ecoo_och,
1104                         FD_OSTDATA_SIZE);
1105                 ecoo->ecoo_oa.o_valid |= OBD_MD_FLHANDLE;
1106
1107                 rc = obd_close (&ec->ec_conn, &ecoo->ecoo_oa,
1108                                 ecoo->ecoo_object->eco_lsm, NULL);
1109
1110                 CDEBUG (D_INFO, "Closed object "LPX64" on disconnect (%d)\n",
1111                         ecoo->ecoo_oa.o_id, rc);
1112
1113                 echo_put_object (ecoo->ecoo_object);
1114                 OBD_FREE (ecoo, sizeof (*ecoo));
1115         }
1116
1117         rc = class_disconnect (conn, 0);
1118         GOTO(out, rc);
1119  out:
1120         class_export_put(exp);
1121         return rc;
1122 }
1123
1124 static struct obd_ops echo_obd_ops = {
1125         o_owner:       THIS_MODULE,
1126         o_setup:       echo_setup,
1127         o_cleanup:     echo_cleanup,
1128         o_iocontrol:   echo_iocontrol,
1129         o_connect:     echo_connect,
1130         o_disconnect:  echo_disconnect
1131 };
1132
1133 int echo_client_init(void)
1134 {
1135         struct lprocfs_static_vars lvars;
1136
1137         lprocfs_init_vars(&lvars);
1138         return class_register_type(&echo_obd_ops, lvars.module_vars,
1139                                    OBD_ECHO_CLIENT_DEVICENAME);
1140 }
1141
1142 void echo_client_cleanup(void)
1143 {
1144         class_unregister_type(OBD_ECHO_CLIENT_DEVICENAME);
1145 }