lustre/obdecho/echo_client.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  */
  32
  33 #define DEBUG_SUBSYSTEM S_ECHO
  34
  35 #include <linux/user_namespace.h>
  36 #include <linux/uidgid.h>
  37
  38 #include <libcfs/libcfs.h>
  39 #include <obd.h>
  40 #include <obd_support.h>
  41 #include <obd_class.h>
  42 #include <lustre_debug.h>
  43 #include <lprocfs_status.h>
  44 #include <cl_object.h>
  45 #include <lustre_fid.h>
  46 #include <lustre_lmv.h>
  47 #include <lustre_acl.h>
  48 #include <uapi/linux/lustre/lustre_ioctl.h>
  49 #include <lustre_net.h>
  50 #ifdef HAVE_SERVER_SUPPORT
  51 # include <md_object.h>
  52
  53 #define ETI_NAME_LEN    20
  54
  55 #endif /* HAVE_SERVER_SUPPORT */
  56
  57 #include "echo_internal.h"
  58
  59 /** \defgroup echo_client Echo Client
  60  * @{
  61  */
  62
  63 /* echo thread key have a CL_THREAD flag, which set cl_env function directly */
  64 #define ECHO_MD_CTX_TAG (LCT_REMEMBER | LCT_MD_THREAD)
  65 #define ECHO_DT_CTX_TAG (LCT_REMEMBER | LCT_DT_THREAD)
  66 #define ECHO_SES_TAG    (LCT_REMEMBER | LCT_SESSION | LCT_SERVER_SESSION)
  67
  68 struct echo_device {
  69         struct cl_device          ed_cl;
  70         struct echo_client_obd   *ed_ec;
  71
  72         struct cl_site            ed_site_myself;
  73         struct lu_site           *ed_site;
  74         struct lu_device         *ed_next;
  75         int                       ed_next_ismd;
  76         struct lu_client_seq     *ed_cl_seq;
  77 #ifdef HAVE_SERVER_SUPPORT
  78         struct local_oid_storage *ed_los;
  79         struct lu_fid             ed_root_fid;
  80 #endif /* HAVE_SERVER_SUPPORT */
  81 };
  82
  83 struct echo_object {
  84         struct cl_object        eo_cl;
  85         struct cl_object_header eo_hdr;
  86         struct echo_device     *eo_dev;
  87         struct list_head        eo_obj_chain;
  88         struct lov_oinfo       *eo_oinfo;
  89         atomic_t                eo_npages;
  90         int                     eo_deleted;
  91 };
  92
  93 struct echo_object_conf {
  94         struct cl_object_conf   eoc_cl;
  95         struct lov_oinfo      **eoc_oinfo;
  96 };
  97
  98 struct echo_page {
  99         struct cl_page_slice    ep_cl;
 100         unsigned long           ep_lock;
 101 };
 102
 103 struct echo_lock {
 104         struct cl_lock_slice    el_cl;
 105         struct list_head        el_chain;
 106         struct echo_object     *el_object;
 107         __u64                   el_cookie;
 108         atomic_t                el_refcount;
 109 };
 110
 111 #ifdef HAVE_SERVER_SUPPORT
 112 static const char echo_md_root_dir_name[] = "ROOT_ECHO";
 113
 114 /**
 115  * In order to use the values of members in struct mdd_device,
 116  * we define an alias structure here.
 117  */
 118 struct echo_md_device {
 119         struct md_device                 emd_md_dev;
 120         struct obd_export               *emd_child_exp;
 121         struct dt_device                *emd_child;
 122         struct dt_device                *emd_bottom;
 123         struct lu_fid                    emd_root_fid;
 124         struct lu_fid                    emd_local_root_fid;
 125 };
 126 #endif /* HAVE_SERVER_SUPPORT */
 127
 128 static int echo_client_setup(const struct lu_env *env,
 129                              struct obd_device *obd,
 130                              struct lustre_cfg *lcfg);
 131 static int echo_client_cleanup(struct obd_device *obd);
 132
 133 /** \defgroup echo_helpers Helper functions
 134  * @{
 135  */
 136 static inline struct echo_device *cl2echo_dev(const struct cl_device *dev)
 137 {
 138         return container_of0(dev, struct echo_device, ed_cl);
 139 }
 140
 141 static inline struct cl_device *echo_dev2cl(struct echo_device *d)
 142 {
 143         return &d->ed_cl;
 144 }
 145
 146 static inline struct echo_device *obd2echo_dev(const struct obd_device *obd)
 147 {
 148         return cl2echo_dev(lu2cl_dev(obd->obd_lu_dev));
 149 }
 150
 151 static inline struct cl_object *echo_obj2cl(struct echo_object *eco)
 152 {
 153         return &eco->eo_cl;
 154 }
 155
 156 static inline struct echo_object *cl2echo_obj(const struct cl_object *o)
 157 {
 158         return container_of(o, struct echo_object, eo_cl);
 159 }
 160
 161 static inline struct echo_page *cl2echo_page(const struct cl_page_slice *s)
 162 {
 163         return container_of(s, struct echo_page, ep_cl);
 164 }
 165
 166 static inline struct echo_lock *cl2echo_lock(const struct cl_lock_slice *s)
 167 {
 168         return container_of(s, struct echo_lock, el_cl);
 169 }
 170
 171 static inline struct cl_lock *echo_lock2cl(const struct echo_lock *ecl)
 172 {
 173         return ecl->el_cl.cls_lock;
 174 }
 175
 176 static struct lu_context_key echo_thread_key;
 177
 178 static inline struct echo_thread_info *echo_env_info(const struct lu_env *env)
 179 {
 180         struct echo_thread_info *info;
 181
 182         info = lu_context_key_get(&env->le_ctx, &echo_thread_key);
 183         LASSERT(info != NULL);
 184         return info;
 185 }
 186
 187 static inline
 188 struct echo_object_conf *cl2echo_conf(const struct cl_object_conf *c)
 189 {
 190         return container_of(c, struct echo_object_conf, eoc_cl);
 191 }
 192
 193 #ifdef HAVE_SERVER_SUPPORT
 194 static inline struct echo_md_device *lu2emd_dev(struct lu_device *d)
 195 {
 196         return container_of0(d, struct echo_md_device, emd_md_dev.md_lu_dev);
 197 }
 198
 199 static inline struct lu_device *emd2lu_dev(struct echo_md_device *d)
 200 {
 201         return &d->emd_md_dev.md_lu_dev;
 202 }
 203
 204 static inline struct seq_server_site *echo_md_seq_site(struct echo_md_device *d)
 205 {
 206         return emd2lu_dev(d)->ld_site->ld_seq_site;
 207 }
 208
 209 static inline struct obd_device *emd2obd_dev(struct echo_md_device *d)
 210 {
 211         return d->emd_md_dev.md_lu_dev.ld_obd;
 212 }
 213 #endif /* HAVE_SERVER_SUPPORT */
 214
 215 /** @} echo_helpers */
 216
 217 static int cl_echo_object_put(struct echo_object *eco);
 218 static int cl_echo_object_brw(struct echo_object *eco, int rw, u64 offset,
 219                               struct page **pages, int npages, int async);
 220
 221 struct echo_thread_info {
 222         struct echo_object_conf eti_conf;
 223         struct lustre_md        eti_md;
 224         struct cl_2queue        eti_queue;
 225         struct cl_io            eti_io;
 226         struct cl_lock          eti_lock;
 227         struct lu_fid           eti_fid;
 228         struct lu_fid           eti_fid2;
 229 #ifdef HAVE_SERVER_SUPPORT
 230         struct md_op_spec       eti_spec;
 231         struct lov_mds_md_v3    eti_lmm;
 232         struct lov_user_md_v3   eti_lum;
 233         struct md_attr          eti_ma;
 234         struct lu_name          eti_lname;
 235         /* per-thread values, can be re-used */
 236         void                    *eti_big_lmm; /* may be vmalloc'd */
 237         int                     eti_big_lmmsize;
 238         char                    eti_name[ETI_NAME_LEN];
 239         struct lu_buf           eti_buf;
 240         /* If we want to test large ACL, then need to enlarge the buffer. */
 241         char                    eti_xattr_buf[LUSTRE_POSIX_ACL_MAX_SIZE_OLD];
 242 #endif
 243 };
 244
 245 /* No session used right now */
 246 struct echo_session_info {
 247         unsigned long dummy;
 248 };
 249
 250 static struct kmem_cache *echo_lock_kmem;
 251 static struct kmem_cache *echo_object_kmem;
 252 static struct kmem_cache *echo_thread_kmem;
 253 static struct kmem_cache *echo_session_kmem;
 254 /* static struct kmem_cache *echo_req_kmem; */
 255
 256 static struct lu_kmem_descr echo_caches[] = {
 257         {
 258                 .ckd_cache = &echo_lock_kmem,
 259                 .ckd_name  = "echo_lock_kmem",
 260                 .ckd_size  = sizeof(struct echo_lock)
 261         },
 262         {
 263                 .ckd_cache = &echo_object_kmem,
 264                 .ckd_name  = "echo_object_kmem",
 265                 .ckd_size  = sizeof(struct echo_object)
 266         },
 267         {
 268                 .ckd_cache = &echo_thread_kmem,
 269                 .ckd_name  = "echo_thread_kmem",
 270                 .ckd_size  = sizeof(struct echo_thread_info)
 271         },
 272         {
 273                 .ckd_cache = &echo_session_kmem,
 274                 .ckd_name  = "echo_session_kmem",
 275                 .ckd_size  = sizeof(struct echo_session_info)
 276         },
 277         {
 278                 .ckd_cache = NULL
 279         }
 280 };
 281
 282 /** \defgroup echo_page Page operations
 283  *
 284  * Echo page operations.
 285  *
 286  * @{
 287  */
 288 static int echo_page_own(const struct lu_env *env,
 289                          const struct cl_page_slice *slice,
 290                          struct cl_io *io, int nonblock)
 291 {
 292         struct echo_page *ep = cl2echo_page(slice);
 293
 294         if (!nonblock) {
 295                 if (test_and_set_bit(0, &ep->ep_lock))
 296                         return -EAGAIN;
 297         } else {
 298                 while (test_and_set_bit(0, &ep->ep_lock))
 299                         wait_on_bit(&ep->ep_lock, 0, TASK_UNINTERRUPTIBLE);
 300         }
 301         return 0;
 302 }
 303
 304 static void echo_page_disown(const struct lu_env *env,
 305                              const struct cl_page_slice *slice,
 306                              struct cl_io *io)
 307 {
 308         struct echo_page *ep = cl2echo_page(slice);
 309
 310         LASSERT(test_bit(0, &ep->ep_lock));
 311         clear_and_wake_up_bit(0, &ep->ep_lock);
 312 }
 313
 314 static void echo_page_discard(const struct lu_env *env,
 315                               const struct cl_page_slice *slice,
 316                               struct cl_io *unused)
 317 {
 318         cl_page_delete(env, slice->cpl_page);
 319 }
 320
 321 static int echo_page_is_vmlocked(const struct lu_env *env,
 322                                  const struct cl_page_slice *slice)
 323 {
 324         if (test_bit(0, &cl2echo_page(slice)->ep_lock))
 325                 return -EBUSY;
 326         return -ENODATA;
 327 }
 328
 329 static void echo_page_completion(const struct lu_env *env,
 330                                  const struct cl_page_slice *slice,
 331                                  int ioret)
 332 {
 333         LASSERT(slice->cpl_page->cp_sync_io != NULL);
 334 }
 335
 336 static void echo_page_fini(const struct lu_env *env,
 337                            struct cl_page_slice *slice,
 338                            struct pagevec *pvec)
 339 {
 340         struct echo_object *eco = cl2echo_obj(slice->cpl_obj);
 341
 342         ENTRY;
 343         atomic_dec(&eco->eo_npages);
 344         put_page(slice->cpl_page->cp_vmpage);
 345         EXIT;
 346 }
 347
 348 static int echo_page_prep(const struct lu_env *env,
 349                           const struct cl_page_slice *slice,
 350                           struct cl_io *unused)
 351 {
 352         return 0;
 353 }
 354
 355 static int echo_page_print(const struct lu_env *env,
 356                            const struct cl_page_slice *slice,
 357                            void *cookie, lu_printer_t printer)
 358 {
 359         struct echo_page *ep = cl2echo_page(slice);
 360
 361         (*printer)(env, cookie, LUSTRE_ECHO_CLIENT_NAME"-page@%p %d vm@%p\n",
 362                    ep, test_bit(0, &ep->ep_lock),
 363                    slice->cpl_page->cp_vmpage);
 364         return 0;
 365 }
 366
 367 static const struct cl_page_operations echo_page_ops = {
 368         .cpo_own           = echo_page_own,
 369         .cpo_disown        = echo_page_disown,
 370         .cpo_discard       = echo_page_discard,
 371         .cpo_fini          = echo_page_fini,
 372         .cpo_print         = echo_page_print,
 373         .cpo_is_vmlocked   = echo_page_is_vmlocked,
 374         .io = {
 375                 [CRT_READ] = {
 376                         .cpo_prep        = echo_page_prep,
 377                         .cpo_completion  = echo_page_completion,
 378                 },
 379                 [CRT_WRITE] = {
 380                         .cpo_prep        = echo_page_prep,
 381                         .cpo_completion  = echo_page_completion,
 382                 }
 383         }
 384 };
 385
 386 /** @} echo_page */
 387
 388 /** \defgroup echo_lock Locking
 389  *
 390  * echo lock operations
 391  *
 392  * @{
 393  */
 394 static void echo_lock_fini(const struct lu_env *env,
 395                            struct cl_lock_slice *slice)
 396 {
 397         struct echo_lock *ecl = cl2echo_lock(slice);
 398
 399         LASSERT(list_empty(&ecl->el_chain));
 400         OBD_SLAB_FREE_PTR(ecl, echo_lock_kmem);
 401 }
 402
 403 static struct cl_lock_operations echo_lock_ops = {
 404         .clo_fini      = echo_lock_fini,
 405 };
 406
 407 /** @} echo_lock */
 408
 409 /** \defgroup echo_cl_ops cl_object operations
 410  *
 411  * operations for cl_object
 412  *
 413  * @{
 414  */
 415 static int echo_page_init(const struct lu_env *env, struct cl_object *obj,
 416                           struct cl_page *page, pgoff_t index)
 417 {
 418         struct echo_page *ep = cl_object_page_slice(obj, page);
 419         struct echo_object *eco = cl2echo_obj(obj);
 420
 421         ENTRY;
 422         get_page(page->cp_vmpage);
 423         /*
 424          * ep_lock is similar to the lock_page() lock, and
 425          * cannot usefully be monitored by lockdep.
 426          * So just use a bit in an "unsigned long" and use the
 427          * wait_on_bit() interface to wait for the bit to be clear.
 428          */
 429         ep->ep_lock = 0;
 430         cl_page_slice_add(page, &ep->ep_cl, obj, index, &echo_page_ops);
 431         atomic_inc(&eco->eo_npages);
 432         RETURN(0);
 433 }
 434
 435 static int echo_io_init(const struct lu_env *env, struct cl_object *obj,
 436                         struct cl_io *io)
 437 {
 438         return 0;
 439 }
 440
 441 static int echo_lock_init(const struct lu_env *env,
 442                           struct cl_object *obj, struct cl_lock *lock,
 443                           const struct cl_io *unused)
 444 {
 445         struct echo_lock *el;
 446
 447         ENTRY;
 448         OBD_SLAB_ALLOC_PTR_GFP(el, echo_lock_kmem, GFP_NOFS);
 449         if (el) {
 450                 cl_lock_slice_add(lock, &el->el_cl, obj, &echo_lock_ops);
 451                 el->el_object = cl2echo_obj(obj);
 452                 INIT_LIST_HEAD(&el->el_chain);
 453                 atomic_set(&el->el_refcount, 0);
 454         }
 455         RETURN(el ? 0 : -ENOMEM);
 456 }
 457
 458 static int echo_conf_set(const struct lu_env *env, struct cl_object *obj,
 459                          const struct cl_object_conf *conf)
 460 {
 461         return 0;
 462 }
 463
 464 static const struct cl_object_operations echo_cl_obj_ops = {
 465         .coo_page_init = echo_page_init,
 466         .coo_lock_init = echo_lock_init,
 467         .coo_io_init   = echo_io_init,
 468         .coo_conf_set  = echo_conf_set
 469 };
 470 /** @} echo_cl_ops */
 471
 472 /** \defgroup echo_lu_ops lu_object operations
 473  *
 474  * operations for echo lu object.
 475  *
 476  * @{
 477  */
 478 static int echo_object_init(const struct lu_env *env, struct lu_object *obj,
 479                             const struct lu_object_conf *conf)
 480 {
 481         struct echo_device *ed         = cl2echo_dev(lu2cl_dev(obj->lo_dev));
 482         struct echo_client_obd *ec     = ed->ed_ec;
 483         struct echo_object *eco        = cl2echo_obj(lu2cl(obj));
 484
 485         ENTRY;
 486         if (ed->ed_next) {
 487                 struct lu_object  *below;
 488                 struct lu_device  *under;
 489
 490                 under = ed->ed_next;
 491                 below = under->ld_ops->ldo_object_alloc(env, obj->lo_header,
 492                                                         under);
 493                 if (!below)
 494                         RETURN(-ENOMEM);
 495                 lu_object_add(obj, below);
 496         }
 497
 498         if (!ed->ed_next_ismd) {
 499                 const struct cl_object_conf *cconf = lu2cl_conf(conf);
 500                 struct echo_object_conf *econf = cl2echo_conf(cconf);
 501
 502                 LASSERT(econf->eoc_oinfo != NULL);
 503
 504                 /*
 505                  * Transfer the oinfo pointer to eco that it won't be
 506                  * freed.
 507                  */
 508                 eco->eo_oinfo = *econf->eoc_oinfo;
 509                 *econf->eoc_oinfo = NULL;
 510         } else {
 511                 eco->eo_oinfo = NULL;
 512         }
 513
 514         eco->eo_dev = ed;
 515         atomic_set(&eco->eo_npages, 0);
 516         cl_object_page_init(lu2cl(obj), sizeof(struct echo_page));
 517
 518         spin_lock(&ec->ec_lock);
 519         list_add_tail(&eco->eo_obj_chain, &ec->ec_objects);
 520         spin_unlock(&ec->ec_lock);
 521
 522         RETURN(0);
 523 }
 524
 525 static void echo_object_delete(const struct lu_env *env, struct lu_object *obj)
 526 {
 527         struct echo_object *eco    = cl2echo_obj(lu2cl(obj));
 528         struct echo_client_obd *ec;
 529
 530         ENTRY;
 531
 532         /* object delete called unconditolally - layer init or not */
 533         if (eco->eo_dev == NULL)
 534                 return;
 535
 536         ec = eco->eo_dev->ed_ec;
 537
 538         LASSERT(atomic_read(&eco->eo_npages) == 0);
 539
 540         spin_lock(&ec->ec_lock);
 541         list_del_init(&eco->eo_obj_chain);
 542         spin_unlock(&ec->ec_lock);
 543
 544         if (eco->eo_oinfo)
 545                 OBD_FREE_PTR(eco->eo_oinfo);
 546 }
 547
 548 static void echo_object_free_rcu(struct rcu_head *head)
 549 {
 550         struct echo_object *eco = container_of(head, struct echo_object,
 551                                                eo_hdr.coh_lu.loh_rcu);
 552
 553         kmem_cache_free(echo_object_kmem, eco);
 554 }
 555
 556 static void echo_object_free(const struct lu_env *env, struct lu_object *obj)
 557 {
 558         struct echo_object *eco    = cl2echo_obj(lu2cl(obj));
 559
 560         ENTRY;
 561
 562         lu_object_fini(obj);
 563         lu_object_header_fini(obj->lo_header);
 564
 565         OBD_FREE_PRE(eco, sizeof(*eco), "slab-freed");
 566         call_rcu(&eco->eo_hdr.coh_lu.loh_rcu, echo_object_free_rcu);
 567         EXIT;
 568 }
 569
 570 static int echo_object_print(const struct lu_env *env, void *cookie,
 571                              lu_printer_t p, const struct lu_object *o)
 572 {
 573         struct echo_object *obj = cl2echo_obj(lu2cl(o));
 574
 575         return (*p)(env, cookie, "echoclient-object@%p", obj);
 576 }
 577
 578 static const struct lu_object_operations echo_lu_obj_ops = {
 579         .loo_object_init      = echo_object_init,
 580         .loo_object_delete    = echo_object_delete,
 581         .loo_object_release   = NULL,
 582         .loo_object_free      = echo_object_free,
 583         .loo_object_print     = echo_object_print,
 584         .loo_object_invariant = NULL
 585 };
 586 /** @} echo_lu_ops */
 587
 588 /** \defgroup echo_lu_dev_ops  lu_device operations
 589  *
 590  * Operations for echo lu device.
 591  *
 592  * @{
 593  */
 594 static struct lu_object *echo_object_alloc(const struct lu_env *env,
 595                                            const struct lu_object_header *hdr,
 596                                            struct lu_device *dev)
 597 {
 598         struct echo_object *eco;
 599         struct lu_object *obj = NULL;
 600
 601         ENTRY;
 602         /* we're the top dev. */
 603         LASSERT(hdr == NULL);
 604         OBD_SLAB_ALLOC_PTR_GFP(eco, echo_object_kmem, GFP_NOFS);
 605         if (eco) {
 606                 struct cl_object_header *hdr = &eco->eo_hdr;
 607
 608                 obj = &echo_obj2cl(eco)->co_lu;
 609                 cl_object_header_init(hdr);
 610                 hdr->coh_page_bufsize = cfs_size_round(sizeof(struct cl_page));
 611
 612                 lu_object_init(obj, &hdr->coh_lu, dev);
 613                 lu_object_add_top(&hdr->coh_lu, obj);
 614
 615                 eco->eo_cl.co_ops = &echo_cl_obj_ops;
 616                 obj->lo_ops       = &echo_lu_obj_ops;
 617         }
 618         RETURN(obj);
 619 }
 620
 621 static struct lu_device_operations echo_device_lu_ops = {
 622         .ldo_object_alloc   = echo_object_alloc,
 623 };
 624
 625 /** @} echo_lu_dev_ops */
 626
 627 /** \defgroup echo_init Setup and teardown
 628  *
 629  * Init and fini functions for echo client.
 630  *
 631  * @{
 632  */
 633 static int echo_site_init(const struct lu_env *env, struct echo_device *ed)
 634 {
 635         struct cl_site *site = &ed->ed_site_myself;
 636         int rc;
 637
 638         /* initialize site */
 639         rc = cl_site_init(site, &ed->ed_cl);
 640         if (rc) {
 641                 CERROR("Cannot initialize site for echo client(%d)\n", rc);
 642                 return rc;
 643         }
 644
 645         rc = lu_site_init_finish(&site->cs_lu);
 646         if (rc) {
 647                 cl_site_fini(site);
 648                 return rc;
 649         }
 650
 651         ed->ed_site = &site->cs_lu;
 652         return 0;
 653 }
 654
 655 static void echo_site_fini(const struct lu_env *env, struct echo_device *ed)
 656 {
 657         if (ed->ed_site) {
 658                 if (!ed->ed_next_ismd)
 659                         lu_site_fini(ed->ed_site);
 660                 ed->ed_site = NULL;
 661         }
 662 }
 663
 664 static void *echo_thread_key_init(const struct lu_context *ctx,
 665                                   struct lu_context_key *key)
 666 {
 667         struct echo_thread_info *info;
 668
 669         OBD_SLAB_ALLOC_PTR_GFP(info, echo_thread_kmem, GFP_NOFS);
 670         if (!info)
 671                 info = ERR_PTR(-ENOMEM);
 672         return info;
 673 }
 674
 675 static void echo_thread_key_fini(const struct lu_context *ctx,
 676                                  struct lu_context_key *key, void *data)
 677 {
 678         struct echo_thread_info *info = data;
 679
 680         OBD_SLAB_FREE_PTR(info, echo_thread_kmem);
 681 }
 682
 683 static struct lu_context_key echo_thread_key = {
 684         .lct_tags = LCT_CL_THREAD,
 685         .lct_init = echo_thread_key_init,
 686         .lct_fini = echo_thread_key_fini,
 687 };
 688
 689 static void *echo_session_key_init(const struct lu_context *ctx,
 690                                   struct lu_context_key *key)
 691 {
 692         struct echo_session_info *session;
 693
 694         OBD_SLAB_ALLOC_PTR_GFP(session, echo_session_kmem, GFP_NOFS);
 695         if (!session)
 696                 session = ERR_PTR(-ENOMEM);
 697         return session;
 698 }
 699
 700 static void echo_session_key_fini(const struct lu_context *ctx,
 701                                   struct lu_context_key *key, void *data)
 702 {
 703         struct echo_session_info *session = data;
 704
 705         OBD_SLAB_FREE_PTR(session, echo_session_kmem);
 706 }
 707
 708 static struct lu_context_key echo_session_key = {
 709         .lct_tags = LCT_SESSION,
 710         .lct_init = echo_session_key_init,
 711         .lct_fini = echo_session_key_fini,
 712 };
 713
 714 LU_TYPE_INIT_FINI(echo, &echo_thread_key, &echo_session_key);
 715
 716 #ifdef HAVE_SERVER_SUPPORT
 717 # define ECHO_SEQ_WIDTH 0xffffffff
 718 static int echo_fid_init(struct echo_device *ed, char *obd_name,
 719                          struct seq_server_site *ss)
 720 {
 721         char *prefix;
 722         int rc;
 723
 724         ENTRY;
 725         OBD_ALLOC_PTR(ed->ed_cl_seq);
 726         if (!ed->ed_cl_seq)
 727                 RETURN(-ENOMEM);
 728
 729         OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
 730         if (!prefix)
 731                 GOTO(out_free_seq, rc = -ENOMEM);
 732
 733         snprintf(prefix, MAX_OBD_NAME + 5, "srv-%s", obd_name);
 734
 735         /* Init client side sequence-manager */
 736         rc = seq_client_init(ed->ed_cl_seq, NULL,
 737                              LUSTRE_SEQ_METADATA,
 738                              prefix, ss->ss_server_seq);
 739         ed->ed_cl_seq->lcs_width = ECHO_SEQ_WIDTH;
 740         OBD_FREE(prefix, MAX_OBD_NAME + 5);
 741         if (rc)
 742                 GOTO(out_free_seq, rc);
 743
 744         RETURN(0);
 745
 746 out_free_seq:
 747         OBD_FREE_PTR(ed->ed_cl_seq);
 748         ed->ed_cl_seq = NULL;
 749         RETURN(rc);
 750 }
 751
 752 static int echo_fid_fini(struct obd_device *obd)
 753 {
 754         struct echo_device *ed = obd2echo_dev(obd);
 755
 756         ENTRY;
 757         if (ed->ed_cl_seq) {
 758                 seq_client_fini(ed->ed_cl_seq);
 759                 OBD_FREE_PTR(ed->ed_cl_seq);
 760                 ed->ed_cl_seq = NULL;
 761         }
 762
 763         RETURN(0);
 764 }
 765
 766 static void echo_ed_los_fini(const struct lu_env *env, struct echo_device *ed)
 767 {
 768         ENTRY;
 769         if (ed != NULL && ed->ed_next_ismd && ed->ed_los != NULL) {
 770                 local_oid_storage_fini(env, ed->ed_los);
 771                 ed->ed_los = NULL;
 772         }
 773 }
 774
 775 static int
 776 echo_md_local_file_create(const struct lu_env *env, struct echo_md_device *emd,
 777                           struct local_oid_storage *los,
 778                           const struct lu_fid *pfid, const char *name,
 779                           __u32 mode, struct lu_fid *fid)
 780 {
 781         struct dt_object        *parent = NULL;
 782         struct dt_object        *dto = NULL;
 783         int                      rc = 0;
 784
 785         ENTRY;
 786         LASSERT(!fid_is_zero(pfid));
 787         parent = dt_locate(env, emd->emd_bottom, pfid);
 788         if (unlikely(IS_ERR(parent)))
 789                 RETURN(PTR_ERR(parent));
 790
 791         /* create local file with @fid */
 792         dto = local_file_find_or_create_with_fid(env, emd->emd_bottom, fid,
 793                                                  parent, name, mode);
 794         if (IS_ERR(dto))
 795                 GOTO(out_put, rc = PTR_ERR(dto));
 796
 797         *fid = *lu_object_fid(&dto->do_lu);
 798         /*
 799          * since stack is not fully set up the local_storage uses own stack
 800          * and we should drop its object from cache
 801          */
 802         dt_object_put_nocache(env, dto);
 803
 804         EXIT;
 805 out_put:
 806         dt_object_put(env, parent);
 807         RETURN(rc);
 808 }
 809
 810 static int
 811 echo_md_root_get(const struct lu_env *env, struct echo_md_device *emd,
 812                  struct echo_device *ed)
 813 {
 814         struct lu_fid fid;
 815         int rc = 0;
 816
 817         ENTRY;
 818         /* Setup local dirs */
 819         fid.f_seq = FID_SEQ_LOCAL_NAME;
 820         fid.f_oid = 1;
 821         fid.f_ver = 0;
 822         rc = local_oid_storage_init(env, emd->emd_bottom, &fid, &ed->ed_los);
 823         if (rc != 0)
 824                 RETURN(rc);
 825
 826         lu_echo_root_fid(&fid);
 827         if (echo_md_seq_site(emd)->ss_node_id == 0) {
 828                 rc = echo_md_local_file_create(env, emd, ed->ed_los,
 829                                                &emd->emd_local_root_fid,
 830                                                echo_md_root_dir_name, S_IFDIR |
 831                                                S_IRUGO | S_IWUSR | S_IXUGO,
 832                                                &fid);
 833                 if (rc != 0) {
 834                         CERROR("%s: create md echo root fid failed: rc = %d\n",
 835                                emd2obd_dev(emd)->obd_name, rc);
 836                         GOTO(out_los, rc);
 837                 }
 838         }
 839         ed->ed_root_fid = fid;
 840
 841         RETURN(0);
 842 out_los:
 843         echo_ed_los_fini(env, ed);
 844
 845         RETURN(rc);
 846 }
 847 #endif /* HAVE_SERVER_SUPPORT */
 848
 849 static struct lu_device *echo_device_alloc(const struct lu_env *env,
 850                                            struct lu_device_type *t,
 851                                            struct lustre_cfg *cfg)
 852 {
 853         struct lu_device   *next;
 854         struct echo_device *ed;
 855         struct cl_device   *cd;
 856         struct obd_device  *obd = NULL; /* to keep compiler happy */
 857         struct obd_device  *tgt;
 858         const char *tgt_type_name;
 859         int rc;
 860         int cleanup = 0;
 861
 862         ENTRY;
 863         OBD_ALLOC_PTR(ed);
 864         if (!ed)
 865                 GOTO(out, rc = -ENOMEM);
 866
 867         cleanup = 1;
 868         cd = &ed->ed_cl;
 869         rc = cl_device_init(cd, t);
 870         if (rc)
 871                 GOTO(out, rc);
 872
 873         cd->cd_lu_dev.ld_ops = &echo_device_lu_ops;
 874
 875         cleanup = 2;
 876         obd = class_name2obd(lustre_cfg_string(cfg, 0));
 877         LASSERT(obd != NULL);
 878         LASSERT(env != NULL);
 879
 880         tgt = class_name2obd(lustre_cfg_string(cfg, 1));
 881         if (!tgt) {
 882                 CERROR("Can not find tgt device %s\n",
 883                         lustre_cfg_string(cfg, 1));
 884                 GOTO(out, rc = -ENODEV);
 885         }
 886
 887         next = tgt->obd_lu_dev;
 888
 889         if (strcmp(tgt->obd_type->typ_name, LUSTRE_MDT_NAME) == 0) {
 890                 ed->ed_next_ismd = 1;
 891         } else if (strcmp(tgt->obd_type->typ_name, LUSTRE_OST_NAME) == 0 ||
 892                    strcmp(tgt->obd_type->typ_name, LUSTRE_OSC_NAME) == 0) {
 893                 ed->ed_next_ismd = 0;
 894                 rc = echo_site_init(env, ed);
 895                 if (rc)
 896                         GOTO(out, rc);
 897         } else {
 898                 GOTO(out, rc = -EINVAL);
 899         }
 900
 901         cleanup = 3;
 902
 903         rc = echo_client_setup(env, obd, cfg);
 904         if (rc)
 905                 GOTO(out, rc);
 906
 907         ed->ed_ec = &obd->u.echo_client;
 908         cleanup = 4;
 909
 910         if (ed->ed_next_ismd) {
 911 #ifdef HAVE_SERVER_SUPPORT
 912                 /* Suppose to connect to some Metadata layer */
 913                 struct lu_site          *ls = NULL;
 914                 struct lu_device        *ld = NULL;
 915                 struct md_device        *md = NULL;
 916                 struct echo_md_device   *emd = NULL;
 917                 int                      found = 0;
 918
 919                 if (!next) {
 920                         CERROR("%s is not lu device type!\n",
 921                                lustre_cfg_string(cfg, 1));
 922                         GOTO(out, rc = -EINVAL);
 923                 }
 924
 925                 tgt_type_name = lustre_cfg_string(cfg, 2);
 926                 if (!tgt_type_name) {
 927                         CERROR("%s no type name for echo %s setup\n",
 928                                 lustre_cfg_string(cfg, 1),
 929                                 tgt->obd_type->typ_name);
 930                         GOTO(out, rc = -EINVAL);
 931                 }
 932
 933                 ls = next->ld_site;
 934
 935                 spin_lock(&ls->ls_ld_lock);
 936                 list_for_each_entry(ld, &ls->ls_ld_linkage, ld_linkage) {
 937                         if (strcmp(ld->ld_type->ldt_name, tgt_type_name) == 0) {
 938                                 found = 1;
 939                                 break;
 940                         }
 941                 }
 942                 spin_unlock(&ls->ls_ld_lock);
 943
 944                 if (found == 0) {
 945                         CERROR("%s is not lu device type!\n",
 946                                lustre_cfg_string(cfg, 1));
 947                         GOTO(out, rc = -EINVAL);
 948                 }
 949
 950                 next = ld;
 951                 /* For MD echo client, it will use the site in MDS stack */
 952                 ed->ed_site = ls;
 953                 ed->ed_cl.cd_lu_dev.ld_site = ls;
 954                 rc = echo_fid_init(ed, obd->obd_name, lu_site2seq(ls));
 955                 if (rc) {
 956                         CERROR("echo fid init error %d\n", rc);
 957                         GOTO(out, rc);
 958                 }
 959
 960                 md = lu2md_dev(next);
 961                 emd = lu2emd_dev(&md->md_lu_dev);
 962                 rc = echo_md_root_get(env, emd, ed);
 963                 if (rc != 0) {
 964                         CERROR("%s: get root error: rc = %d\n",
 965                                 emd2obd_dev(emd)->obd_name, rc);
 966                         GOTO(out, rc);
 967                 }
 968 #else /* !HAVE_SERVER_SUPPORT */
 969                 CERROR(
 970                        "Local operations are NOT supported on client side. Only remote operations are supported. Metadata client must be run on server side.\n");
 971                 GOTO(out, rc = -EOPNOTSUPP);
 972 #endif /* HAVE_SERVER_SUPPORT */
 973         } else {
 974                 /*
 975                  * if echo client is to be stacked upon ost device, the next is
 976                  * NULL since ost is not a clio device so far
 977                  */
 978                 if (next != NULL && !lu_device_is_cl(next))
 979                         next = NULL;
 980
 981                 tgt_type_name = tgt->obd_type->typ_name;
 982                 if (next) {
 983                         LASSERT(next != NULL);
 984                         if (next->ld_site)
 985                                 GOTO(out, rc = -EBUSY);
 986
 987                         next->ld_site = ed->ed_site;
 988                         rc = next->ld_type->ldt_ops->ldto_device_init(env, next,
 989                                                         next->ld_type->ldt_name,
 990                                                         NULL);
 991                         if (rc)
 992                                 GOTO(out, rc);
 993                 } else {
 994                         LASSERT(strcmp(tgt_type_name, LUSTRE_OST_NAME) == 0);
 995                 }
 996         }
 997
 998         ed->ed_next = next;
 999         RETURN(&cd->cd_lu_dev);
1000 out:
1001         switch (cleanup) {
1002         case 4: {
1003                 int rc2;
1004
1005                 rc2 = echo_client_cleanup(obd);
1006                 if (rc2)
1007                         CERROR("Cleanup obd device %s error(%d)\n",
1008                                obd->obd_name, rc2);
1009         }
1010         /* fallthrough */
1011
1012         case 3:
1013                 echo_site_fini(env, ed);
1014                 /* fallthrough */
1015         case 2:
1016                 cl_device_fini(&ed->ed_cl);
1017                 /* fallthrough */
1018         case 1:
1019                 OBD_FREE_PTR(ed);
1020                 /* fallthrough */
1021         case 0:
1022         default:
1023                 break;
1024         }
1025         return ERR_PTR(rc);
1026 }
1027
1028 static int echo_device_init(const struct lu_env *env, struct lu_device *d,
1029                             const char *name, struct lu_device *next)
1030 {
1031         LBUG();
1032         return 0;
1033 }
1034
1035 static struct lu_device *echo_device_fini(const struct lu_env *env,
1036                                           struct lu_device *d)
1037 {
1038         struct echo_device *ed = cl2echo_dev(lu2cl_dev(d));
1039         struct lu_device *next = ed->ed_next;
1040
1041         while (next && !ed->ed_next_ismd)
1042                 next = next->ld_type->ldt_ops->ldto_device_fini(env, next);
1043         return NULL;
1044 }
1045
1046 static void echo_lock_release(const struct lu_env *env,
1047                               struct echo_lock *ecl,
1048                               int still_used)
1049 {
1050         struct cl_lock *clk = echo_lock2cl(ecl);
1051
1052         cl_lock_release(env, clk);
1053 }
1054
1055 static struct lu_device *echo_device_free(const struct lu_env *env,
1056                                           struct lu_device *d)
1057 {
1058         struct echo_device     *ed   = cl2echo_dev(lu2cl_dev(d));
1059         struct echo_client_obd *ec   = ed->ed_ec;
1060         struct echo_object     *eco;
1061         struct lu_device       *next = ed->ed_next;
1062
1063         CDEBUG(D_INFO, "echo device:%p is going to be freed, next = %p\n",
1064                ed, next);
1065
1066         lu_site_purge(env, ed->ed_site, -1);
1067
1068         /*
1069          * check if there are objects still alive.
1070          * It shouldn't have any object because lu_site_purge would cleanup
1071          * all of cached objects. Anyway, probably the echo device is being
1072          * parallelly accessed.
1073          */
1074         spin_lock(&ec->ec_lock);
1075         list_for_each_entry(eco, &ec->ec_objects, eo_obj_chain)
1076                 eco->eo_deleted = 1;
1077         spin_unlock(&ec->ec_lock);
1078
1079         /* purge again */
1080         lu_site_purge(env, ed->ed_site, -1);
1081
1082         CDEBUG(D_INFO,
1083                "Waiting for the reference of echo object to be dropped\n");
1084
1085         /* Wait for the last reference to be dropped. */
1086         spin_lock(&ec->ec_lock);
1087         while (!list_empty(&ec->ec_objects)) {
1088                 spin_unlock(&ec->ec_lock);
1089                 CERROR(
1090                        "echo_client still has objects at cleanup time, wait for 1 second\n");
1091                 schedule_timeout_uninterruptible(cfs_time_seconds(1));
1092                 lu_site_purge(env, ed->ed_site, -1);
1093                 spin_lock(&ec->ec_lock);
1094         }
1095         spin_unlock(&ec->ec_lock);
1096
1097         LASSERT(list_empty(&ec->ec_locks));
1098
1099         CDEBUG(D_INFO, "No object exists, exiting...\n");
1100
1101         echo_client_cleanup(d->ld_obd);
1102 #ifdef HAVE_SERVER_SUPPORT
1103         echo_fid_fini(d->ld_obd);
1104         echo_ed_los_fini(env, ed);
1105 #endif
1106         while (next && !ed->ed_next_ismd)
1107                 next = next->ld_type->ldt_ops->ldto_device_free(env, next);
1108
1109         LASSERT(ed->ed_site == d->ld_site);
1110         echo_site_fini(env, ed);
1111         cl_device_fini(&ed->ed_cl);
1112         OBD_FREE_PTR(ed);
1113
1114         cl_env_cache_purge(~0);
1115
1116         return NULL;
1117 }
1118
1119 static const struct lu_device_type_operations echo_device_type_ops = {
1120         .ldto_init = echo_type_init,
1121         .ldto_fini = echo_type_fini,
1122
1123         .ldto_start = echo_type_start,
1124         .ldto_stop  = echo_type_stop,
1125
1126         .ldto_device_alloc = echo_device_alloc,
1127         .ldto_device_free  = echo_device_free,
1128         .ldto_device_init  = echo_device_init,
1129         .ldto_device_fini  = echo_device_fini
1130 };
1131
1132 static struct lu_device_type echo_device_type = {
1133         .ldt_tags     = LU_DEVICE_CL,
1134         .ldt_name     = LUSTRE_ECHO_CLIENT_NAME,
1135         .ldt_ops      = &echo_device_type_ops,
1136         .ldt_ctx_tags = LCT_CL_THREAD | LCT_MD_THREAD | LCT_DT_THREAD,
1137 };
1138 /** @} echo_init */
1139
1140 /** \defgroup echo_exports Exported operations
1141  *
1142  * exporting functions to echo client
1143  *
1144  * @{
1145  */
1146
1147 /* Interfaces to echo client obd device */
1148 static struct echo_object *
1149 cl_echo_object_find(struct echo_device *d, const struct ost_id *oi)
1150 {
1151         struct lu_env *env;
1152         struct echo_thread_info *info;
1153         struct echo_object_conf *conf;
1154         struct echo_object *eco;
1155         struct cl_object *obj;
1156         struct lov_oinfo *oinfo = NULL;
1157         struct lu_fid *fid;
1158         __u16  refcheck;
1159         int rc;
1160
1161         ENTRY;
1162         LASSERTF(ostid_id(oi) != 0, DOSTID"\n", POSTID(oi));
1163         LASSERTF(ostid_seq(oi) == FID_SEQ_ECHO, DOSTID"\n", POSTID(oi));
1164
1165         /* Never return an object if the obd is to be freed. */
1166         if (echo_dev2cl(d)->cd_lu_dev.ld_obd->obd_stopping)
1167                 RETURN(ERR_PTR(-ENODEV));
1168
1169         env = cl_env_get(&refcheck);
1170         if (IS_ERR(env))
1171                 RETURN((void *)env);
1172
1173         info = echo_env_info(env);
1174         conf = &info->eti_conf;
1175         if (d->ed_next) {
1176                 OBD_ALLOC_PTR(oinfo);
1177                 if (!oinfo)
1178                         GOTO(out, eco = ERR_PTR(-ENOMEM));
1179
1180                 oinfo->loi_oi = *oi;
1181                 conf->eoc_cl.u.coc_oinfo = oinfo;
1182         }
1183
1184         /*
1185          * If echo_object_init() is successful then ownership of oinfo
1186          * is transferred to the object.
1187          */
1188         conf->eoc_oinfo = &oinfo;
1189
1190         fid = &info->eti_fid;
1191         rc = ostid_to_fid(fid, oi, 0);
1192         if (rc != 0)
1193                 GOTO(out, eco = ERR_PTR(rc));
1194
1195         /*
1196          * In the function below, .hs_keycmp resolves to
1197          * lu_obj_hop_keycmp()
1198          */
1199         /* coverity[overrun-buffer-val] */
1200         obj = cl_object_find(env, echo_dev2cl(d), fid, &conf->eoc_cl);
1201         if (IS_ERR(obj))
1202                 GOTO(out, eco = (void *)obj);
1203
1204         eco = cl2echo_obj(obj);
1205         if (eco->eo_deleted) {
1206                 cl_object_put(env, obj);
1207                 eco = ERR_PTR(-EAGAIN);
1208         }
1209
1210 out:
1211         if (oinfo)
1212                 OBD_FREE_PTR(oinfo);
1213
1214         cl_env_put(env, &refcheck);
1215         RETURN(eco);
1216 }
1217
1218 static int cl_echo_object_put(struct echo_object *eco)
1219 {
1220         struct lu_env *env;
1221         struct cl_object *obj = echo_obj2cl(eco);
1222         __u16  refcheck;
1223
1224         ENTRY;
1225         env = cl_env_get(&refcheck);
1226         if (IS_ERR(env))
1227                 RETURN(PTR_ERR(env));
1228
1229         /* an external function to kill an object? */
1230         if (eco->eo_deleted) {
1231                 struct lu_object_header *loh = obj->co_lu.lo_header;
1232
1233                 LASSERT(&eco->eo_hdr == luh2coh(loh));
1234                 set_bit(LU_OBJECT_HEARD_BANSHEE, &loh->loh_flags);
1235         }
1236
1237         cl_object_put(env, obj);
1238         cl_env_put(env, &refcheck);
1239         RETURN(0);
1240 }
1241
1242 static int cl_echo_enqueue0(struct lu_env *env, struct echo_object *eco,
1243                             u64 start, u64 end, int mode,
1244                             __u64 *cookie, __u32 enqflags)
1245 {
1246         struct cl_io *io;
1247         struct cl_lock *lck;
1248         struct cl_object *obj;
1249         struct cl_lock_descr *descr;
1250         struct echo_thread_info *info;
1251         int rc = -ENOMEM;
1252
1253         ENTRY;
1254         info = echo_env_info(env);
1255         io = &info->eti_io;
1256         lck = &info->eti_lock;
1257         obj = echo_obj2cl(eco);
1258
1259         memset(lck, 0, sizeof(*lck));
1260         descr = &lck->cll_descr;
1261         descr->cld_obj   = obj;
1262         descr->cld_start = cl_index(obj, start);
1263         descr->cld_end   = cl_index(obj, end);
1264         descr->cld_mode  = mode == LCK_PW ? CLM_WRITE : CLM_READ;
1265         descr->cld_enq_flags = enqflags;
1266         io->ci_obj = obj;
1267
1268         rc = cl_lock_request(env, io, lck);
1269         if (rc == 0) {
1270                 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
1271                 struct echo_lock *el;
1272
1273                 el = cl2echo_lock(cl_lock_at(lck, &echo_device_type));
1274                 spin_lock(&ec->ec_lock);
1275                 if (list_empty(&el->el_chain)) {
1276                         list_add(&el->el_chain, &ec->ec_locks);
1277                         el->el_cookie = ++ec->ec_unique;
1278                 }
1279                 atomic_inc(&el->el_refcount);
1280                 *cookie = el->el_cookie;
1281                 spin_unlock(&ec->ec_lock);
1282         }
1283         RETURN(rc);
1284 }
1285
1286 static int cl_echo_cancel0(struct lu_env *env, struct echo_device *ed,
1287                            __u64 cookie)
1288 {
1289         struct echo_client_obd *ec = ed->ed_ec;
1290         struct echo_lock *ecl = NULL;
1291         struct list_head *el;
1292         int found = 0, still_used = 0;
1293
1294         ENTRY;
1295         LASSERT(ec != NULL);
1296         spin_lock(&ec->ec_lock);
1297         list_for_each(el, &ec->ec_locks) {
1298                 ecl = list_entry(el, struct echo_lock, el_chain);
1299                 CDEBUG(D_INFO, "ecl: %p, cookie: %#llx\n", ecl, ecl->el_cookie);
1300                 found = (ecl->el_cookie == cookie);
1301                 if (found) {
1302                         if (atomic_dec_and_test(&ecl->el_refcount))
1303                                 list_del_init(&ecl->el_chain);
1304                         else
1305                                 still_used = 1;
1306                         break;
1307                 }
1308         }
1309         spin_unlock(&ec->ec_lock);
1310
1311         if (!found)
1312                 RETURN(-ENOENT);
1313
1314         echo_lock_release(env, ecl, still_used);
1315         RETURN(0);
1316 }
1317
1318 static void echo_commit_callback(const struct lu_env *env, struct cl_io *io,
1319                                  struct pagevec *pvec)
1320 {
1321         struct echo_thread_info *info;
1322         struct cl_2queue        *queue;
1323         int i = 0;
1324
1325         info = echo_env_info(env);
1326         LASSERT(io == &info->eti_io);
1327
1328         queue = &info->eti_queue;
1329
1330         for (i = 0; i < pagevec_count(pvec); i++) {
1331                 struct page *vmpage = pvec->pages[i];
1332                 struct cl_page *page = (struct cl_page *)vmpage->private;
1333
1334                 cl_page_list_add(&queue->c2_qout, page);
1335         }
1336 }
1337
1338 static int cl_echo_object_brw(struct echo_object *eco, int rw, u64 offset,
1339                               struct page **pages, int npages, int async)
1340 {
1341         struct lu_env           *env;
1342         struct echo_thread_info *info;
1343         struct cl_object        *obj = echo_obj2cl(eco);
1344         struct echo_device      *ed  = eco->eo_dev;
1345         struct cl_2queue        *queue;
1346         struct cl_io            *io;
1347         struct cl_page          *clp;
1348         struct lustre_handle    lh = { 0 };
1349         int page_size = cl_page_size(obj);
1350         int rc;
1351         int i;
1352         __u16 refcheck;
1353
1354         ENTRY;
1355         LASSERT((offset & ~PAGE_MASK) == 0);
1356         LASSERT(ed->ed_next != NULL);
1357         env = cl_env_get(&refcheck);
1358         if (IS_ERR(env))
1359                 RETURN(PTR_ERR(env));
1360
1361         info    = echo_env_info(env);
1362         io      = &info->eti_io;
1363         queue   = &info->eti_queue;
1364
1365         cl_2queue_init(queue);
1366
1367         io->ci_ignore_layout = 1;
1368         rc = cl_io_init(env, io, CIT_MISC, obj);
1369         if (rc < 0)
1370                 GOTO(out, rc);
1371         LASSERT(rc == 0);
1372
1373         rc = cl_echo_enqueue0(env, eco, offset,
1374                               offset + npages * PAGE_SIZE - 1,
1375                               rw == READ ? LCK_PR : LCK_PW, &lh.cookie,
1376                               CEF_NEVER);
1377         if (rc < 0)
1378                 GOTO(error_lock, rc);
1379
1380         for (i = 0; i < npages; i++) {
1381                 LASSERT(pages[i]);
1382                 clp = cl_page_find(env, obj, cl_index(obj, offset),
1383                                    pages[i], CPT_TRANSIENT);
1384                 if (IS_ERR(clp)) {
1385                         rc = PTR_ERR(clp);
1386                         break;
1387                 }
1388                 LASSERT(clp->cp_type == CPT_TRANSIENT);
1389
1390                 rc = cl_page_own(env, io, clp);
1391                 if (rc) {
1392                         LASSERT(clp->cp_state == CPS_FREEING);
1393                         cl_page_put(env, clp);
1394                         break;
1395                 }
1396
1397                 cl_2queue_add(queue, clp);
1398
1399                 /*
1400                  * drop the reference count for cl_page_find, so that the page
1401                  * will be freed in cl_2queue_fini.
1402                  */
1403                 cl_page_put(env, clp);
1404                 cl_page_clip(env, clp, 0, page_size);
1405
1406                 offset += page_size;
1407         }
1408
1409         if (rc == 0) {
1410                 enum cl_req_type typ = rw == READ ? CRT_READ : CRT_WRITE;
1411
1412                 async = async && (typ == CRT_WRITE);
1413                 if (async)
1414                         rc = cl_io_commit_async(env, io, &queue->c2_qin,
1415                                                 0, PAGE_SIZE,
1416                                                 echo_commit_callback);
1417                 else
1418                         rc = cl_io_submit_sync(env, io, typ, queue, 0);
1419                 CDEBUG(D_INFO, "echo_client %s write returns %d\n",
1420                        async ? "async" : "sync", rc);
1421         }
1422
1423         cl_echo_cancel0(env, ed, lh.cookie);
1424         EXIT;
1425 error_lock:
1426         cl_2queue_discard(env, io, queue);
1427         cl_2queue_disown(env, io, queue);
1428         cl_2queue_fini(env, queue);
1429         cl_io_fini(env, io);
1430 out:
1431         cl_env_put(env, &refcheck);
1432         return rc;
1433 }
1434 /** @} echo_exports */
1435
1436 static u64 last_object_id;
1437
1438 #ifdef HAVE_SERVER_SUPPORT
1439 static inline void echo_md_build_name(struct lu_name *lname, char *name,
1440                                       __u64 id)
1441 {
1442         snprintf(name, ETI_NAME_LEN, "%llu", id);
1443         lname->ln_name = name;
1444         lname->ln_namelen = strlen(name);
1445 }
1446
1447 /* similar to mdt_attr_get_complex */
1448 static int echo_big_lmm_get(const struct lu_env *env, struct md_object *o,
1449                             struct md_attr *ma)
1450 {
1451         struct echo_thread_info *info = echo_env_info(env);
1452         int rc;
1453
1454         ENTRY;
1455
1456         LASSERT(ma->ma_lmm_size > 0);
1457
1458         LASSERT(ma->ma_need & (MA_LOV | MA_LMV));
1459         if (ma->ma_need & MA_LOV)
1460                 rc = mo_xattr_get(env, o, &LU_BUF_NULL, XATTR_NAME_LOV);
1461         else
1462                 rc = mo_xattr_get(env, o, &LU_BUF_NULL, XATTR_NAME_LMV);
1463
1464         if (rc < 0)
1465                 RETURN(rc);
1466
1467         /* big_lmm may need to be grown */
1468         if (info->eti_big_lmmsize < rc) {
1469                 int size = size_roundup_power2(rc);
1470
1471                 if (info->eti_big_lmmsize > 0) {
1472                         /* free old buffer */
1473                         LASSERT(info->eti_big_lmm);
1474                         OBD_FREE_LARGE(info->eti_big_lmm,
1475                                        info->eti_big_lmmsize);
1476                         info->eti_big_lmm = NULL;
1477                         info->eti_big_lmmsize = 0;
1478                 }
1479
1480                 OBD_ALLOC_LARGE(info->eti_big_lmm, size);
1481                 if (!info->eti_big_lmm)
1482                         RETURN(-ENOMEM);
1483                 info->eti_big_lmmsize = size;
1484         }
1485         LASSERT(info->eti_big_lmmsize >= rc);
1486
1487         info->eti_buf.lb_buf = info->eti_big_lmm;
1488         info->eti_buf.lb_len = info->eti_big_lmmsize;
1489         if (ma->ma_need & MA_LOV)
1490                 rc = mo_xattr_get(env, o, &info->eti_buf, XATTR_NAME_LOV);
1491         else
1492                 rc = mo_xattr_get(env, o, &info->eti_buf, XATTR_NAME_LMV);
1493         if (rc < 0)
1494                 RETURN(rc);
1495
1496         if (ma->ma_need & MA_LOV)
1497                 ma->ma_valid |= MA_LOV;
1498         else
1499                 ma->ma_valid |= MA_LMV;
1500
1501         ma->ma_lmm = info->eti_big_lmm;
1502         ma->ma_lmm_size = rc;
1503
1504         RETURN(0);
1505 }
1506
1507 static int echo_attr_get_complex(const struct lu_env *env,
1508                                  struct md_object *next,
1509                                  struct md_attr *ma)
1510 {
1511         struct echo_thread_info *info = echo_env_info(env);
1512         struct lu_buf           *buf = &info->eti_buf;
1513         umode_t                  mode = lu_object_attr(&next->mo_lu);
1514         int                      rc = 0, rc2;
1515
1516         ENTRY;
1517
1518         ma->ma_valid = 0;
1519
1520         if (ma->ma_need & MA_INODE) {
1521                 rc = mo_attr_get(env, next, ma);
1522                 if (rc)
1523                         GOTO(out, rc);
1524                 ma->ma_valid |= MA_INODE;
1525         }
1526
1527         if ((ma->ma_need & MA_LOV) && (S_ISREG(mode) || S_ISDIR(mode))) {
1528                 LASSERT(ma->ma_lmm_size > 0);
1529                 buf->lb_buf = ma->ma_lmm;
1530                 buf->lb_len = ma->ma_lmm_size;
1531                 rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_LOV);
1532                 if (rc2 > 0) {
1533                         ma->ma_lmm_size = rc2;
1534                         ma->ma_valid |= MA_LOV;
1535                 } else if (rc2 == -ENODATA) {
1536                         /* no LOV EA */
1537                         ma->ma_lmm_size = 0;
1538                 } else if (rc2 == -ERANGE) {
1539                         rc2 = echo_big_lmm_get(env, next, ma);
1540                         if (rc2 < 0)
1541                                 GOTO(out, rc = rc2);
1542                 } else {
1543                         GOTO(out, rc = rc2);
1544                 }
1545         }
1546
1547         if ((ma->ma_need & MA_LMV) && S_ISDIR(mode)) {
1548                 LASSERT(ma->ma_lmm_size > 0);
1549                 buf->lb_buf = ma->ma_lmm;
1550                 buf->lb_len = ma->ma_lmm_size;
1551                 rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_LMV);
1552                 if (rc2 > 0) {
1553                         ma->ma_lmm_size = rc2;
1554                         ma->ma_valid |= MA_LMV;
1555                 } else if (rc2 == -ENODATA) {
1556                         /* no LMV EA */
1557                         ma->ma_lmm_size = 0;
1558                 } else if (rc2 == -ERANGE) {
1559                         rc2 = echo_big_lmm_get(env, next, ma);
1560                         if (rc2 < 0)
1561                                 GOTO(out, rc = rc2);
1562                 } else {
1563                         GOTO(out, rc = rc2);
1564                 }
1565         }
1566
1567 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
1568         if ((ma->ma_need & MA_ACL_DEF) && S_ISDIR(mode)) {
1569                 buf->lb_buf = ma->ma_acl;
1570                 buf->lb_len = ma->ma_acl_size;
1571                 rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_ACL_DEFAULT);
1572                 if (rc2 > 0) {
1573                         ma->ma_acl_size = rc2;
1574                         ma->ma_valid |= MA_ACL_DEF;
1575                 } else if (rc2 == -ENODATA) {
1576                         /* no ACLs */
1577                         ma->ma_acl_size = 0;
1578                 } else {
1579                         GOTO(out, rc = rc2);
1580                 }
1581         }
1582 #endif
1583 out:
1584         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = %#llx ma_lmm=%p\n",
1585                rc, ma->ma_valid, ma->ma_lmm);
1586         RETURN(rc);
1587 }
1588
1589 static int
1590 echo_md_create_internal(const struct lu_env *env, struct echo_device *ed,
1591                         struct md_object *parent, struct lu_fid *fid,
1592                         struct lu_name *lname, struct md_op_spec *spec,
1593                         struct md_attr *ma)
1594 {
1595         struct lu_object        *ec_child, *child;
1596         struct lu_device        *ld = ed->ed_next;
1597         struct echo_thread_info *info = echo_env_info(env);
1598         struct lu_fid           *fid2 = &info->eti_fid2;
1599         struct lu_object_conf    conf = { .loc_flags = LOC_F_NEW };
1600         int                      rc;
1601
1602         ENTRY;
1603
1604         rc = mdo_lookup(env, parent, lname, fid2, spec);
1605         if (rc == 0)
1606                 return -EEXIST;
1607         else if (rc != -ENOENT)
1608                 return rc;
1609
1610         ec_child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev,
1611                                      fid, &conf);
1612         if (IS_ERR(ec_child)) {
1613                 CERROR("Can not find the child "DFID": rc = %ld\n", PFID(fid),
1614                         PTR_ERR(ec_child));
1615                 RETURN(PTR_ERR(ec_child));
1616         }
1617
1618         child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1619         if (!child) {
1620                 CERROR("Can not locate the child "DFID"\n", PFID(fid));
1621                 GOTO(out_put, rc = -EINVAL);
1622         }
1623
1624         CDEBUG(D_RPCTRACE, "Start creating object "DFID" %s %p\n",
1625                PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
1626
1627         /*
1628          * Do not perform lookup sanity check. We know that name does not exist.
1629          */
1630         spec->sp_cr_lookup = 0;
1631         rc = mdo_create(env, parent, lname, lu2md(child), spec, ma);
1632         if (rc) {
1633                 CERROR("Can not create child "DFID": rc = %d\n", PFID(fid), rc);
1634                 GOTO(out_put, rc);
1635         }
1636         CDEBUG(D_RPCTRACE, "End creating object "DFID" %s %p rc  = %d\n",
1637                PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent, rc);
1638         EXIT;
1639 out_put:
1640         lu_object_put(env, ec_child);
1641         return rc;
1642 }
1643
1644 static int echo_set_lmm_size(const struct lu_env *env, struct lu_device *ld,
1645                              struct md_attr *ma)
1646 {
1647         struct echo_thread_info *info = echo_env_info(env);
1648
1649         if (strcmp(ld->ld_type->ldt_name, LUSTRE_MDD_NAME)) {
1650                 ma->ma_lmm = (void *)&info->eti_lmm;
1651                 ma->ma_lmm_size = sizeof(info->eti_lmm);
1652         } else {
1653                 LASSERT(info->eti_big_lmmsize);
1654                 ma->ma_lmm = info->eti_big_lmm;
1655                 ma->ma_lmm_size = info->eti_big_lmmsize;
1656         }
1657
1658         return 0;
1659 }
1660
1661 static int
1662 echo_md_dir_stripe_choose(const struct lu_env *env, struct echo_device *ed,
1663                           struct lu_object *obj, const char *name,
1664                           unsigned int namelen, __u64 id,
1665                           struct lu_object **new_parent)
1666 {
1667         struct echo_thread_info *info = echo_env_info(env);
1668         struct md_attr          *ma = &info->eti_ma;
1669         struct lmv_mds_md_v1    *lmv;
1670         struct lu_device        *ld = ed->ed_next;
1671         unsigned int            idx;
1672         struct lu_name          tmp_ln_name;
1673         struct lu_fid           stripe_fid;
1674         struct lu_object        *stripe_obj;
1675         int                     rc;
1676
1677         LASSERT(obj != NULL);
1678         LASSERT(S_ISDIR(obj->lo_header->loh_attr));
1679
1680         memset(ma, 0, sizeof(*ma));
1681         echo_set_lmm_size(env, ld, ma);
1682         ma->ma_need = MA_LMV;
1683         rc = echo_attr_get_complex(env, lu2md(obj), ma);
1684         if (rc) {
1685                 CERROR("Can not getattr child "DFID": rc = %d\n",
1686                         PFID(lu_object_fid(obj)), rc);
1687                 return rc;
1688         }
1689
1690         if (!(ma->ma_valid & MA_LMV)) {
1691                 *new_parent = obj;
1692                 return 0;
1693         }
1694
1695         lmv = (struct lmv_mds_md_v1 *)ma->ma_lmm;
1696         if (!lmv_is_sane(lmv)) {
1697                 rc = -EINVAL;
1698                 CERROR("Invalid mds md magic %x "DFID": rc = %d\n",
1699                        le32_to_cpu(lmv->lmv_magic), PFID(lu_object_fid(obj)),
1700                        rc);
1701                 return rc;
1702         }
1703
1704         if (name) {
1705                 tmp_ln_name.ln_name = name;
1706                 tmp_ln_name.ln_namelen = namelen;
1707         } else {
1708                 LASSERT(id != -1);
1709                 echo_md_build_name(&tmp_ln_name, info->eti_name, id);
1710         }
1711
1712         idx = lmv_name_to_stripe_index(lmv, tmp_ln_name.ln_name,
1713                                        tmp_ln_name.ln_namelen);
1714
1715         LASSERT(idx < le32_to_cpu(lmv->lmv_stripe_count));
1716         fid_le_to_cpu(&stripe_fid, &lmv->lmv_stripe_fids[idx]);
1717
1718         stripe_obj = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, &stripe_fid,
1719                                        NULL);
1720         if (IS_ERR(stripe_obj)) {
1721                 rc = PTR_ERR(stripe_obj);
1722                 CERROR("Can not find the parent "DFID": rc = %d\n",
1723                        PFID(&stripe_fid), rc);
1724                 return rc;
1725         }
1726
1727         *new_parent = lu_object_locate(stripe_obj->lo_header, ld->ld_type);
1728         if (!*new_parent) {
1729                 lu_object_put(env, stripe_obj);
1730                 RETURN(-ENXIO);
1731         }
1732
1733         return rc;
1734 }
1735
1736 static int echo_create_md_object(const struct lu_env *env,
1737                                  struct echo_device *ed,
1738                                  struct lu_object *ec_parent,
1739                                  struct lu_fid *fid,
1740                                  char *name, int namelen,
1741                                   __u64 id, __u32 mode, int count,
1742                                  int stripe_count, int stripe_offset)
1743 {
1744         struct lu_object *parent;
1745         struct lu_object *new_parent;
1746         struct echo_thread_info *info = echo_env_info(env);
1747         struct lu_name *lname = &info->eti_lname;
1748         struct md_op_spec *spec = &info->eti_spec;
1749         struct md_attr *ma = &info->eti_ma;
1750         struct lu_device *ld = ed->ed_next;
1751         int rc = 0;
1752         int i;
1753
1754         ENTRY;
1755
1756         if (!ec_parent)
1757                 return -1;
1758         parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1759         if (!parent)
1760                 RETURN(-ENXIO);
1761
1762         rc = echo_md_dir_stripe_choose(env, ed, parent, name, namelen,
1763                                        id, &new_parent);
1764         if (rc != 0)
1765                 RETURN(rc);
1766
1767         LASSERT(new_parent != NULL);
1768         memset(ma, 0, sizeof(*ma));
1769         memset(spec, 0, sizeof(*spec));
1770         echo_set_lmm_size(env, ld, ma);
1771         if (stripe_count != 0) {
1772                 spec->sp_cr_flags |= MDS_FMODE_WRITE;
1773                 if (stripe_count != -1) {
1774                         if (S_ISDIR(mode)) {
1775                                 struct lmv_user_md *lmu;
1776
1777                                 lmu = (struct lmv_user_md *)&info->eti_lum;
1778                                 lmu->lum_magic = LMV_USER_MAGIC;
1779                                 lmu->lum_stripe_offset = stripe_offset;
1780                                 lmu->lum_stripe_count = stripe_count;
1781                                 lmu->lum_hash_type = LMV_HASH_TYPE_FNV_1A_64;
1782                                 spec->u.sp_ea.eadata = lmu;
1783                                 spec->u.sp_ea.eadatalen = sizeof(*lmu);
1784                         } else {
1785                                 struct lov_user_md_v3 *lum = &info->eti_lum;
1786
1787                                 lum->lmm_magic = LOV_USER_MAGIC_V3;
1788                                 lum->lmm_stripe_count = stripe_count;
1789                                 lum->lmm_stripe_offset = stripe_offset;
1790                                 lum->lmm_pattern = LOV_PATTERN_NONE;
1791                                 spec->u.sp_ea.eadata = lum;
1792                                 spec->u.sp_ea.eadatalen = sizeof(*lum);
1793                         }
1794                         spec->sp_cr_flags |= MDS_OPEN_HAS_EA;
1795                 }
1796         }
1797
1798         ma->ma_attr.la_mode = mode;
1799         ma->ma_attr.la_valid = LA_CTIME | LA_MODE;
1800         ma->ma_attr.la_ctime = ktime_get_real_seconds();
1801
1802         if (name) {
1803                 lname->ln_name = name;
1804                 lname->ln_namelen = namelen;
1805                 /* If name is specified, only create one object by name */
1806                 rc = echo_md_create_internal(env, ed, lu2md(new_parent), fid,
1807                                              lname, spec, ma);
1808                 GOTO(out_put, rc);
1809         }
1810
1811         /* Create multiple object sequenced by id */
1812         for (i = 0; i < count; i++) {
1813                 char *tmp_name = info->eti_name;
1814
1815                 echo_md_build_name(lname, tmp_name, id);
1816
1817                 rc = echo_md_create_internal(env, ed, lu2md(new_parent),
1818                                              fid, lname, spec, ma);
1819                 if (rc) {
1820                         CERROR("Can not create child %s: rc = %d\n", tmp_name,
1821                                 rc);
1822                         break;
1823                 }
1824                 id++;
1825                 fid->f_oid++;
1826         }
1827
1828 out_put:
1829         if (new_parent != parent)
1830                 lu_object_put(env, new_parent);
1831
1832         RETURN(rc);
1833 }
1834
1835 static struct lu_object *echo_md_lookup(const struct lu_env *env,
1836                                         struct echo_device *ed,
1837                                         struct md_object *parent,
1838                                         struct lu_name *lname)
1839 {
1840         struct echo_thread_info *info = echo_env_info(env);
1841         struct lu_fid *fid = &info->eti_fid;
1842         struct lu_object *child;
1843         int rc;
1844
1845         ENTRY;
1846         CDEBUG(D_INFO, "lookup %s in parent "DFID" %p\n", lname->ln_name,
1847                PFID(fid), parent);
1848
1849         rc = mdo_lookup(env, parent, lname, fid, NULL);
1850         if (rc) {
1851                 CERROR("lookup %s: rc = %d\n", lname->ln_name, rc);
1852                 RETURN(ERR_PTR(rc));
1853         }
1854
1855         /*
1856          * In the function below, .hs_keycmp resolves to
1857          * lu_obj_hop_keycmp()
1858          */
1859         /* coverity[overrun-buffer-val] */
1860         child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, fid, NULL);
1861
1862         RETURN(child);
1863 }
1864
1865 static int echo_setattr_object(const struct lu_env *env,
1866                                struct echo_device *ed,
1867                                struct lu_object *ec_parent,
1868                                __u64 id, int count)
1869 {
1870         struct lu_object *parent;
1871         struct lu_object *new_parent;
1872         struct echo_thread_info *info = echo_env_info(env);
1873         struct lu_name *lname = &info->eti_lname;
1874         char *name = info->eti_name;
1875         struct lu_device *ld = ed->ed_next;
1876         struct lu_buf *buf = &info->eti_buf;
1877         int rc = 0;
1878         int i;
1879
1880         ENTRY;
1881
1882         if (!ec_parent)
1883                 return -1;
1884         parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1885         if (!parent)
1886                 RETURN(-ENXIO);
1887
1888         rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
1889                                        &new_parent);
1890         if (rc != 0)
1891                 RETURN(rc);
1892
1893         for (i = 0; i < count; i++) {
1894                 struct lu_object *ec_child, *child;
1895
1896                 echo_md_build_name(lname, name, id);
1897
1898                 ec_child = echo_md_lookup(env, ed, lu2md(new_parent), lname);
1899                 if (IS_ERR(ec_child)) {
1900                         rc = PTR_ERR(ec_child);
1901                         CERROR("Can't find child %s: rc = %d\n",
1902                                 lname->ln_name, rc);
1903                         break;
1904                 }
1905
1906                 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1907                 if (!child) {
1908                         CERROR("Can not locate the child %s\n", lname->ln_name);
1909                         lu_object_put(env, ec_child);
1910                         rc = -EINVAL;
1911                         break;
1912                 }
1913
1914                 CDEBUG(D_RPCTRACE, "Start setattr object "DFID"\n",
1915                        PFID(lu_object_fid(child)));
1916
1917                 buf->lb_buf = info->eti_xattr_buf;
1918                 buf->lb_len = sizeof(info->eti_xattr_buf);
1919
1920                 sprintf(name, "%s.test1", XATTR_USER_PREFIX);
1921                 rc = mo_xattr_set(env, lu2md(child), buf, name,
1922                                   LU_XATTR_CREATE);
1923                 if (rc < 0) {
1924                         CERROR("Can not setattr child "DFID": rc = %d\n",
1925                                 PFID(lu_object_fid(child)), rc);
1926                         lu_object_put(env, ec_child);
1927                         break;
1928                 }
1929                 CDEBUG(D_RPCTRACE, "End setattr object "DFID"\n",
1930                        PFID(lu_object_fid(child)));
1931                 id++;
1932                 lu_object_put(env, ec_child);
1933         }
1934
1935         if (new_parent != parent)
1936                 lu_object_put(env, new_parent);
1937
1938         RETURN(rc);
1939 }
1940
1941 static int echo_getattr_object(const struct lu_env *env,
1942                                struct echo_device *ed,
1943                                struct lu_object *ec_parent,
1944                                __u64 id, int count)
1945 {
1946         struct lu_object *parent;
1947         struct lu_object *new_parent;
1948         struct echo_thread_info *info = echo_env_info(env);
1949         struct lu_name *lname = &info->eti_lname;
1950         char *name = info->eti_name;
1951         struct md_attr *ma = &info->eti_ma;
1952         struct lu_device *ld = ed->ed_next;
1953         int rc = 0;
1954         int i;
1955
1956         ENTRY;
1957
1958         if (!ec_parent)
1959                 return -1;
1960         parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1961         if (!parent)
1962                 RETURN(-ENXIO);
1963
1964         rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
1965                                        &new_parent);
1966         if (rc != 0)
1967                 RETURN(rc);
1968
1969         memset(ma, 0, sizeof(*ma));
1970         ma->ma_need |= MA_INODE | MA_LOV | MA_PFID | MA_HSM | MA_ACL_DEF;
1971         ma->ma_acl = info->eti_xattr_buf;
1972         ma->ma_acl_size = sizeof(info->eti_xattr_buf);
1973
1974         for (i = 0; i < count; i++) {
1975                 struct lu_object *ec_child, *child;
1976
1977                 ma->ma_valid = 0;
1978                 echo_md_build_name(lname, name, id);
1979                 echo_set_lmm_size(env, ld, ma);
1980
1981                 ec_child = echo_md_lookup(env, ed, lu2md(new_parent), lname);
1982                 if (IS_ERR(ec_child)) {
1983                         CERROR("Can't find child %s: rc = %ld\n",
1984                                lname->ln_name, PTR_ERR(ec_child));
1985                         RETURN(PTR_ERR(ec_child));
1986                 }
1987
1988                 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1989                 if (!child) {
1990                         CERROR("Can not locate the child %s\n", lname->ln_name);
1991                         lu_object_put(env, ec_child);
1992                         RETURN(-EINVAL);
1993                 }
1994
1995                 CDEBUG(D_RPCTRACE, "Start getattr object "DFID"\n",
1996                        PFID(lu_object_fid(child)));
1997                 rc = echo_attr_get_complex(env, lu2md(child), ma);
1998                 if (rc) {
1999                         CERROR("Can not getattr child "DFID": rc = %d\n",
2000                                 PFID(lu_object_fid(child)), rc);
2001                         lu_object_put(env, ec_child);
2002                         break;
2003                 }
2004                 CDEBUG(D_RPCTRACE, "End getattr object "DFID"\n",
2005                        PFID(lu_object_fid(child)));
2006                 id++;
2007                 lu_object_put(env, ec_child);
2008         }
2009
2010         if (new_parent != parent)
2011                 lu_object_put(env, new_parent);
2012
2013         RETURN(rc);
2014 }
2015
2016 static int echo_lookup_object(const struct lu_env *env,
2017                               struct echo_device *ed,
2018                               struct lu_object *ec_parent,
2019                               __u64 id, int count)
2020 {
2021         struct lu_object *parent;
2022         struct lu_object *new_parent;
2023         struct echo_thread_info *info = echo_env_info(env);
2024         struct lu_name *lname = &info->eti_lname;
2025         char *name = info->eti_name;
2026         struct lu_fid *fid = &info->eti_fid;
2027         struct lu_device *ld = ed->ed_next;
2028         int rc = 0;
2029         int i;
2030
2031         if (!ec_parent)
2032                 return -1;
2033         parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
2034         if (!parent)
2035                 return -ENXIO;
2036
2037         rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
2038                                        &new_parent);
2039         if (rc != 0)
2040                 RETURN(rc);
2041
2042         /*prepare the requests*/
2043         for (i = 0; i < count; i++) {
2044                 echo_md_build_name(lname, name, id);
2045
2046                 CDEBUG(D_RPCTRACE, "Start lookup object "DFID" %s %p\n",
2047                        PFID(lu_object_fid(new_parent)), lname->ln_name,
2048                        new_parent);
2049
2050                 rc = mdo_lookup(env, lu2md(new_parent), lname, fid, NULL);
2051                 if (rc) {
2052                         CERROR("Can not lookup child %s: rc = %d\n", name, rc);
2053                         break;
2054                 }
2055
2056                 CDEBUG(D_RPCTRACE, "End lookup object "DFID" %s %p\n",
2057                        PFID(lu_object_fid(new_parent)), lname->ln_name,
2058                        new_parent);
2059
2060                 id++;
2061         }
2062
2063         if (new_parent != parent)
2064                 lu_object_put(env, new_parent);
2065
2066         return rc;
2067 }
2068
2069 static int echo_md_destroy_internal(const struct lu_env *env,
2070                                     struct echo_device *ed,
2071                                     struct md_object *parent,
2072                                     struct lu_name *lname,
2073                                     struct md_attr *ma)
2074 {
2075         struct lu_device   *ld = ed->ed_next;
2076         struct lu_object   *ec_child;
2077         struct lu_object   *child;
2078         int                 rc;
2079
2080         ENTRY;
2081
2082         ec_child = echo_md_lookup(env, ed, parent, lname);
2083         if (IS_ERR(ec_child)) {
2084                 CERROR("Can't find child %s: rc = %ld\n", lname->ln_name,
2085                         PTR_ERR(ec_child));
2086                 RETURN(PTR_ERR(ec_child));
2087         }
2088
2089         child = lu_object_locate(ec_child->lo_header, ld->ld_type);
2090         if (!child) {
2091                 CERROR("Can not locate the child %s\n", lname->ln_name);
2092                 GOTO(out_put, rc = -EINVAL);
2093         }
2094
2095         if (lu_object_remote(child)) {
2096                 CERROR("Can not destroy remote object %s: rc = %d\n",
2097                        lname->ln_name, -EPERM);
2098                 GOTO(out_put, rc = -EPERM);
2099         }
2100         CDEBUG(D_RPCTRACE, "Start destroy object "DFID" %s %p\n",
2101                PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
2102
2103         rc = mdo_unlink(env, parent, lu2md(child), lname, ma, 0);
2104         if (rc) {
2105                 CERROR("Can not unlink child %s: rc = %d\n",
2106                         lname->ln_name, rc);
2107                 GOTO(out_put, rc);
2108         }
2109         CDEBUG(D_RPCTRACE, "End destroy object "DFID" %s %p\n",
2110                PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
2111 out_put:
2112         lu_object_put(env, ec_child);
2113         return rc;
2114 }
2115
2116 static int echo_destroy_object(const struct lu_env *env,
2117                                struct echo_device *ed,
2118                                struct lu_object *ec_parent,
2119                                char *name, int namelen,
2120                                __u64 id, __u32 mode,
2121                                int count)
2122 {
2123         struct echo_thread_info *info = echo_env_info(env);
2124         struct lu_name          *lname = &info->eti_lname;
2125         struct md_attr          *ma = &info->eti_ma;
2126         struct lu_device        *ld = ed->ed_next;
2127         struct lu_object        *parent;
2128         struct lu_object        *new_parent;
2129         int                      rc = 0;
2130         int                      i;
2131
2132         ENTRY;
2133         parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
2134         if (!parent)
2135                 RETURN(-EINVAL);
2136
2137         rc = echo_md_dir_stripe_choose(env, ed, parent, name, namelen,
2138                                        id, &new_parent);
2139         if (rc != 0)
2140                 RETURN(rc);
2141
2142         memset(ma, 0, sizeof(*ma));
2143         ma->ma_attr.la_mode = mode;
2144         ma->ma_attr.la_valid = LA_CTIME;
2145         ma->ma_attr.la_ctime = ktime_get_real_seconds();
2146         ma->ma_need = MA_INODE;
2147         ma->ma_valid = 0;
2148
2149         if (name) {
2150                 lname->ln_name = name;
2151                 lname->ln_namelen = namelen;
2152                 rc = echo_md_destroy_internal(env, ed, lu2md(new_parent), lname,
2153                                               ma);
2154                 GOTO(out_put, rc);
2155         }
2156
2157         /*prepare the requests*/
2158         for (i = 0; i < count; i++) {
2159                 char *tmp_name = info->eti_name;
2160
2161                 ma->ma_valid = 0;
2162                 echo_md_build_name(lname, tmp_name, id);
2163
2164                 rc = echo_md_destroy_internal(env, ed, lu2md(new_parent), lname,
2165                                               ma);
2166                 if (rc) {
2167                         CERROR("Can not unlink child %s: rc = %d\n", name, rc);
2168                         break;
2169                 }
2170                 id++;
2171         }
2172
2173 out_put:
2174         if (new_parent != parent)
2175                 lu_object_put(env, new_parent);
2176
2177         RETURN(rc);
2178 }
2179
2180 static struct lu_object *echo_resolve_path(const struct lu_env *env,
2181                                            struct echo_device *ed, char *path,
2182                                            int path_len)
2183 {
2184         struct lu_device        *ld = ed->ed_next;
2185         struct echo_thread_info *info = echo_env_info(env);
2186         struct lu_fid           *fid = &info->eti_fid;
2187         struct lu_name          *lname = &info->eti_lname;
2188         struct lu_object        *parent = NULL;
2189         struct lu_object        *child = NULL;
2190         int                      rc = 0;
2191
2192         ENTRY;
2193         *fid = ed->ed_root_fid;
2194
2195         /*
2196          * In the function below, .hs_keycmp resolves to
2197          * lu_obj_hop_keycmp()
2198          */
2199         /* coverity[overrun-buffer-val] */
2200         parent = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, fid, NULL);
2201         if (IS_ERR(parent)) {
2202                 CERROR("Can not find the parent "DFID": rc = %ld\n",
2203                         PFID(fid), PTR_ERR(parent));
2204                 RETURN(parent);
2205         }
2206
2207         while (1) {
2208                 struct lu_object *ld_parent;
2209                 char *e;
2210
2211                 e = strsep(&path, "/");
2212                 if (!e)
2213                         break;
2214
2215                 if (e[0] == 0) {
2216                         if (!path || path[0] == '\0')
2217                                 break;
2218                         continue;
2219                 }
2220
2221                 lname->ln_name = e;
2222                 lname->ln_namelen = strlen(e);
2223
2224                 ld_parent = lu_object_locate(parent->lo_header, ld->ld_type);
2225                 if (!ld_parent) {
2226                         lu_object_put(env, parent);
2227                         rc = -EINVAL;
2228                         break;
2229                 }
2230
2231                 child = echo_md_lookup(env, ed, lu2md(ld_parent), lname);
2232                 lu_object_put(env, parent);
2233                 if (IS_ERR(child)) {
2234                         rc = (int)PTR_ERR(child);
2235                         CERROR("lookup %s under parent "DFID": rc = %d\n",
2236                                 lname->ln_name, PFID(lu_object_fid(ld_parent)),
2237                                 rc);
2238                         break;
2239                 }
2240                 parent = child;
2241         }
2242         if (rc)
2243                 RETURN(ERR_PTR(rc));
2244
2245         RETURN(parent);
2246 }
2247
2248 static void echo_ucred_init(struct lu_env *env)
2249 {
2250         struct lu_ucred *ucred = lu_ucred(env);
2251
2252         ucred->uc_valid = UCRED_INVALID;
2253
2254         ucred->uc_suppgids[0] = -1;
2255         ucred->uc_suppgids[1] = -1;
2256
2257         ucred->uc_uid = ucred->uc_o_uid  =
2258                                 from_kuid(&init_user_ns, current_uid());
2259         ucred->uc_gid = ucred->uc_o_gid  =
2260                                 from_kgid(&init_user_ns, current_gid());
2261         ucred->uc_fsuid = ucred->uc_o_fsuid =
2262                                 from_kuid(&init_user_ns, current_fsuid());
2263         ucred->uc_fsgid = ucred->uc_o_fsgid =
2264                                 from_kgid(&init_user_ns, current_fsgid());
2265         ucred->uc_cap = cfs_curproc_cap_pack();
2266
2267         /* remove fs privilege for non-root user. */
2268         if (ucred->uc_fsuid)
2269                 ucred->uc_cap &= ~CFS_CAP_FS_MASK;
2270         ucred->uc_valid = UCRED_NEW;
2271 }
2272
2273 static void echo_ucred_fini(struct lu_env *env)
2274 {
2275         struct lu_ucred *ucred = lu_ucred(env);
2276
2277         ucred->uc_valid = UCRED_INIT;
2278 }
2279
2280 static int echo_md_handler(struct echo_device *ed, int command,
2281                            char *path, int path_len, __u64 id, int count,
2282                            struct obd_ioctl_data *data)
2283 {
2284         struct echo_thread_info *info;
2285         struct lu_device *ld = ed->ed_next;
2286         struct lu_env *env;
2287         __u16 refcheck;
2288         struct lu_object *parent;
2289         char *name = NULL;
2290         int namelen = data->ioc_plen2;
2291         int rc = 0;
2292
2293         ENTRY;
2294         if (!ld) {
2295                 CERROR("MD echo client is not being initialized properly\n");
2296                 RETURN(-EINVAL);
2297         }
2298
2299         if (strcmp(ld->ld_type->ldt_name, LUSTRE_MDD_NAME)) {
2300                 CERROR("Only support MDD layer right now!\n");
2301                 RETURN(-EINVAL);
2302         }
2303
2304         env = cl_env_get(&refcheck);
2305         if (IS_ERR(env))
2306                 RETURN(PTR_ERR(env));
2307
2308         rc = lu_env_refill_by_tags(env, ECHO_MD_CTX_TAG, ECHO_SES_TAG);
2309         if (rc != 0)
2310                 GOTO(out_env, rc);
2311
2312         /* init big_lmm buffer */
2313         info = echo_env_info(env);
2314         LASSERT(info->eti_big_lmm == NULL);
2315         OBD_ALLOC_LARGE(info->eti_big_lmm, MIN_MD_SIZE);
2316         if (!info->eti_big_lmm)
2317                 GOTO(out_env, rc = -ENOMEM);
2318         info->eti_big_lmmsize = MIN_MD_SIZE;
2319
2320         parent = echo_resolve_path(env, ed, path, path_len);
2321         if (IS_ERR(parent)) {
2322                 CERROR("Can not resolve the path %s: rc = %ld\n", path,
2323                         PTR_ERR(parent));
2324                 GOTO(out_free, rc = PTR_ERR(parent));
2325         }
2326
2327         if (namelen > 0) {
2328                 OBD_ALLOC(name, namelen + 1);
2329                 if (!name)
2330                         GOTO(out_put, rc = -ENOMEM);
2331                 if (copy_from_user(name, data->ioc_pbuf2, namelen))
2332                         GOTO(out_name, rc = -EFAULT);
2333         }
2334
2335         echo_ucred_init(env);
2336
2337         switch (command) {
2338         case ECHO_MD_CREATE:
2339         case ECHO_MD_MKDIR: {
2340                 struct echo_thread_info *info = echo_env_info(env);
2341                 __u32 mode = data->ioc_obdo2.o_mode;
2342                 struct lu_fid *fid = &info->eti_fid;
2343                 int stripe_count = (int)data->ioc_obdo2.o_misc;
2344                 int stripe_index = (int)data->ioc_obdo2.o_stripe_idx;
2345
2346                 rc = ostid_to_fid(fid, &data->ioc_obdo1.o_oi, 0);
2347                 if (rc != 0)
2348                         break;
2349
2350                 /*
2351                  * In the function below, .hs_keycmp resolves to
2352                  * lu_obj_hop_keycmp()
2353                  */
2354                 /* coverity[overrun-buffer-val] */
2355                 rc = echo_create_md_object(env, ed, parent, fid, name, namelen,
2356                                            id, mode, count, stripe_count,
2357                                            stripe_index);
2358                 break;
2359         }
2360         case ECHO_MD_DESTROY:
2361         case ECHO_MD_RMDIR: {
2362                 __u32 mode = data->ioc_obdo2.o_mode;
2363
2364                 rc = echo_destroy_object(env, ed, parent, name, namelen,
2365                                          id, mode, count);
2366                 break;
2367         }
2368         case ECHO_MD_LOOKUP:
2369                 rc = echo_lookup_object(env, ed, parent, id, count);
2370                 break;
2371         case ECHO_MD_GETATTR:
2372                 rc = echo_getattr_object(env, ed, parent, id, count);
2373                 break;
2374         case ECHO_MD_SETATTR:
2375                 rc = echo_setattr_object(env, ed, parent, id, count);
2376                 break;
2377         default:
2378                 CERROR("unknown command %d\n", command);
2379                 rc = -EINVAL;
2380                 break;
2381         }
2382         echo_ucred_fini(env);
2383
2384 out_name:
2385         if (name)
2386                 OBD_FREE(name, namelen + 1);
2387 out_put:
2388         lu_object_put(env, parent);
2389 out_free:
2390         LASSERT(info->eti_big_lmm);
2391         OBD_FREE_LARGE(info->eti_big_lmm, info->eti_big_lmmsize);
2392         info->eti_big_lmm = NULL;
2393         info->eti_big_lmmsize = 0;
2394 out_env:
2395         cl_env_put(env, &refcheck);
2396         return rc;
2397 }
2398 #endif /* HAVE_SERVER_SUPPORT */
2399
2400 static int echo_create_object(const struct lu_env *env, struct echo_device *ed,
2401                               struct obdo *oa)
2402 {
2403         struct echo_object      *eco;
2404         struct echo_client_obd  *ec = ed->ed_ec;
2405         int created = 0;
2406         int rc;
2407
2408         ENTRY;
2409         if (!(oa->o_valid & OBD_MD_FLID) ||
2410             !(oa->o_valid & OBD_MD_FLGROUP) ||
2411             !fid_seq_is_echo(ostid_seq(&oa->o_oi))) {
2412                 CERROR("invalid oid "DOSTID"\n", POSTID(&oa->o_oi));
2413                 RETURN(-EINVAL);
2414         }
2415
2416         if (ostid_id(&oa->o_oi) == 0) {
2417                 rc = ostid_set_id(&oa->o_oi, ++last_object_id);
2418                 if (rc)
2419                         GOTO(failed, rc);
2420         }
2421
2422         rc = obd_create(env, ec->ec_exp, oa);
2423         if (rc != 0) {
2424                 CERROR("Cannot create objects: rc = %d\n", rc);
2425                 GOTO(failed, rc);
2426         }
2427
2428         created = 1;
2429
2430         oa->o_valid |= OBD_MD_FLID;
2431
2432         eco = cl_echo_object_find(ed, &oa->o_oi);
2433         if (IS_ERR(eco))
2434                 GOTO(failed, rc = PTR_ERR(eco));
2435         cl_echo_object_put(eco);
2436
2437         CDEBUG(D_INFO, "oa oid "DOSTID"\n", POSTID(&oa->o_oi));
2438         EXIT;
2439
2440 failed:
2441         if (created && rc != 0)
2442                 obd_destroy(env, ec->ec_exp, oa);
2443
2444         if (rc != 0)
2445                 CERROR("create object failed with: rc = %d\n", rc);
2446
2447         return rc;
2448 }
2449
2450 static int echo_get_object(struct echo_object **ecop, struct echo_device *ed,
2451                            struct obdo *oa)
2452 {
2453         struct echo_object *eco;
2454         int rc;
2455
2456         ENTRY;
2457         if (!(oa->o_valid & OBD_MD_FLID) ||
2458             !(oa->o_valid & OBD_MD_FLGROUP) ||
2459             ostid_id(&oa->o_oi) == 0) {
2460                 CERROR("invalid oid "DOSTID"\n", POSTID(&oa->o_oi));
2461                 RETURN(-EINVAL);
2462         }
2463
2464         rc = 0;
2465         eco = cl_echo_object_find(ed, &oa->o_oi);
2466         if (!IS_ERR(eco))
2467                 *ecop = eco;
2468         else
2469                 rc = PTR_ERR(eco);
2470
2471         RETURN(rc);
2472 }
2473
2474 static void echo_put_object(struct echo_object *eco)
2475 {
2476         int rc;
2477
2478         rc = cl_echo_object_put(eco);
2479         if (rc)
2480                 CERROR("%s: echo client drop an object failed: rc = %d\n",
2481                        eco->eo_dev->ed_ec->ec_exp->exp_obd->obd_name, rc);
2482 }
2483
2484 static void echo_client_page_debug_setup(struct page *page, int rw, u64 id,
2485                                          u64 offset, u64 count)
2486 {
2487         char *addr;
2488         u64 stripe_off;
2489         u64 stripe_id;
2490         int delta;
2491
2492         /* no partial pages on the client */
2493         LASSERT(count == PAGE_SIZE);
2494
2495         addr = kmap(page);
2496
2497         for (delta = 0; delta < PAGE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2498                 if (rw == OBD_BRW_WRITE) {
2499                         stripe_off = offset + delta;
2500                         stripe_id = id;
2501                 } else {
2502                         stripe_off = 0xdeadbeef00c0ffeeULL;
2503                         stripe_id = 0xdeadbeef00c0ffeeULL;
2504                 }
2505                 block_debug_setup(addr + delta, OBD_ECHO_BLOCK_SIZE,
2506                                   stripe_off, stripe_id);
2507         }
2508
2509         kunmap(page);
2510 }
2511
2512 static int
2513 echo_client_page_debug_check(struct page *page, u64 id, u64 offset, u64 count)
2514 {
2515         u64 stripe_off;
2516         u64 stripe_id;
2517         char *addr;
2518         int delta;
2519         int rc;
2520         int rc2;
2521
2522         /* no partial pages on the client */
2523         LASSERT(count == PAGE_SIZE);
2524
2525         addr = kmap(page);
2526
2527         for (rc = delta = 0; delta < PAGE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2528                 stripe_off = offset + delta;
2529                 stripe_id = id;
2530
2531                 rc2 = block_debug_check("test_brw",
2532                                         addr + delta, OBD_ECHO_BLOCK_SIZE,
2533                                         stripe_off, stripe_id);
2534                 if (rc2 != 0) {
2535                         CERROR("Error in echo object %#llx\n", id);
2536                         rc = rc2;
2537                 }
2538         }
2539
2540         kunmap(page);
2541         return rc;
2542 }
2543
2544 static int echo_client_kbrw(struct echo_device *ed, int rw, struct obdo *oa,
2545                             struct echo_object *eco, u64 offset,
2546                             u64 count, int async)
2547 {
2548         size_t npages;
2549         struct brw_page *pga;
2550         struct brw_page *pgp;
2551         struct page **pages;
2552         u64 off;
2553         size_t i;
2554         int rc;
2555         int verify;
2556         gfp_t gfp_mask;
2557         u32 brw_flags = 0;
2558
2559         ENTRY;
2560         verify = (ostid_id(&oa->o_oi) != ECHO_PERSISTENT_OBJID &&
2561                   (oa->o_valid & OBD_MD_FLFLAGS) != 0 &&
2562                   (oa->o_flags & OBD_FL_DEBUG_CHECK) != 0);
2563
2564         gfp_mask = ((ostid_id(&oa->o_oi) & 2) == 0) ? GFP_KERNEL : GFP_HIGHUSER;
2565
2566         LASSERT(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ);
2567
2568         if ((count & (~PAGE_MASK)) != 0)
2569                 RETURN(-EINVAL);
2570
2571         /* XXX think again with misaligned I/O */
2572         npages = count >> PAGE_SHIFT;
2573
2574         if (rw == OBD_BRW_WRITE)
2575                 brw_flags = OBD_BRW_ASYNC;
2576
2577         OBD_ALLOC(pga, npages * sizeof(*pga));
2578         if (!pga)
2579                 RETURN(-ENOMEM);
2580
2581         OBD_ALLOC(pages, npages * sizeof(*pages));
2582         if (!pages) {
2583                 OBD_FREE(pga, npages * sizeof(*pga));
2584                 RETURN(-ENOMEM);
2585         }
2586
2587         for (i = 0, pgp = pga, off = offset;
2588              i < npages;
2589              i++, pgp++, off += PAGE_SIZE) {
2590
2591                 LASSERT(pgp->pg == NULL);       /* for cleanup */
2592
2593                 rc = -ENOMEM;
2594                 pgp->pg = alloc_page(gfp_mask);
2595                 if (!pgp->pg)
2596                         goto out;
2597
2598                 pages[i] = pgp->pg;
2599                 pgp->count = PAGE_SIZE;
2600                 pgp->off = off;
2601                 pgp->flag = brw_flags;
2602
2603                 if (verify)
2604                         echo_client_page_debug_setup(pgp->pg, rw,
2605                                                      ostid_id(&oa->o_oi), off,
2606                                                      pgp->count);
2607         }
2608
2609         /* brw mode can only be used at client */
2610         LASSERT(ed->ed_next != NULL);
2611         rc = cl_echo_object_brw(eco, rw, offset, pages, npages, async);
2612
2613  out:
2614         if (rc != 0 || rw != OBD_BRW_READ)
2615                 verify = 0;
2616
2617         for (i = 0, pgp = pga; i < npages; i++, pgp++) {
2618                 if (!pgp->pg)
2619                         continue;
2620
2621                 if (verify) {
2622                         int vrc;
2623
2624                         vrc = echo_client_page_debug_check(pgp->pg,
2625                                                            ostid_id(&oa->o_oi),
2626                                                            pgp->off,
2627                                                            pgp->count);
2628                         if (vrc != 0 && rc == 0)
2629                                 rc = vrc;
2630                 }
2631                 __free_page(pgp->pg);
2632         }
2633         OBD_FREE(pga, npages * sizeof(*pga));
2634         OBD_FREE(pages, npages * sizeof(*pages));
2635         RETURN(rc);
2636 }
2637
2638 static int echo_client_prep_commit(const struct lu_env *env,
2639                                    struct obd_export *exp, int rw,
2640                                    struct obdo *oa, struct echo_object *eco,
2641                                    u64 offset, u64 count,
2642                                    u64 batch, int async)
2643 {
2644         struct obd_ioobj ioo;
2645         struct niobuf_local *lnb;
2646         struct niobuf_remote rnb;
2647         u64 off;
2648         u64 npages, tot_pages, apc;
2649         int i, ret = 0, brw_flags = 0;
2650
2651         ENTRY;
2652         if (count <= 0 || (count & ~PAGE_MASK) != 0)
2653                 RETURN(-EINVAL);
2654
2655         apc = npages = batch >> PAGE_SHIFT;
2656         tot_pages = count >> PAGE_SHIFT;
2657
2658         OBD_ALLOC_LARGE(lnb, apc * sizeof(*lnb));
2659         if (!lnb)
2660                 RETURN(-ENOMEM);
2661
2662         if (rw == OBD_BRW_WRITE && async)
2663                 brw_flags |= OBD_BRW_ASYNC;
2664
2665         obdo_to_ioobj(oa, &ioo);
2666
2667         off = offset;
2668
2669         for (; tot_pages > 0; tot_pages -= npages) {
2670                 int lpages;
2671
2672                 if (tot_pages < npages)
2673                         npages = tot_pages;
2674
2675                 rnb.rnb_offset = off;
2676                 rnb.rnb_len = npages * PAGE_SIZE;
2677                 rnb.rnb_flags = brw_flags;
2678                 ioo.ioo_bufcnt = 1;
2679                 off += npages * PAGE_SIZE;
2680
2681                 lpages = npages;
2682                 ret = obd_preprw(env, rw, exp, oa, 1, &ioo, &rnb, &lpages, lnb);
2683                 if (ret != 0)
2684                         GOTO(out, ret);
2685
2686                 for (i = 0; i < lpages; i++) {
2687                         struct page *page = lnb[i].lnb_page;
2688
2689                         /* read past eof? */
2690                         if (!page && lnb[i].lnb_rc == 0)
2691                                 continue;
2692
2693                         if (async)
2694                                 lnb[i].lnb_flags |= OBD_BRW_ASYNC;
2695
2696                         if (ostid_id(&oa->o_oi) == ECHO_PERSISTENT_OBJID ||
2697                             (oa->o_valid & OBD_MD_FLFLAGS) == 0 ||
2698                             (oa->o_flags & OBD_FL_DEBUG_CHECK) == 0)
2699                                 continue;
2700
2701                         if (rw == OBD_BRW_WRITE)
2702                                 echo_client_page_debug_setup(page, rw,
2703                                                         ostid_id(&oa->o_oi),
2704                                                         lnb[i].lnb_file_offset,
2705                                                         lnb[i].lnb_len);
2706                         else
2707                                 echo_client_page_debug_check(page,
2708                                                         ostid_id(&oa->o_oi),
2709                                                         lnb[i].lnb_file_offset,
2710                                                         lnb[i].lnb_len);
2711                 }
2712
2713                 ret = obd_commitrw(env, rw, exp, oa, 1, &ioo, &rnb, npages, lnb,
2714                                    ret);
2715                 if (ret != 0)
2716                         break;
2717
2718                 /* Reuse env context. */
2719                 lu_context_exit((struct lu_context *)&env->le_ctx);
2720                 lu_context_enter((struct lu_context *)&env->le_ctx);
2721         }
2722
2723 out:
2724         OBD_FREE_LARGE(lnb, apc * sizeof(*lnb));
2725
2726         RETURN(ret);
2727 }
2728
2729 static int echo_client_brw_ioctl(const struct lu_env *env, int rw,
2730                                  struct obd_export *exp,
2731                                  struct obd_ioctl_data *data)
2732 {
2733         struct obd_device *obd = class_exp2obd(exp);
2734         struct echo_device *ed = obd2echo_dev(obd);
2735         struct echo_client_obd *ec = ed->ed_ec;
2736         struct obdo *oa = &data->ioc_obdo1;
2737         struct echo_object *eco;
2738         int rc;
2739         int async = 0;
2740         long test_mode;
2741
2742         ENTRY;
2743         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
2744
2745         rc = echo_get_object(&eco, ed, oa);
2746         if (rc)
2747                 RETURN(rc);
2748
2749         oa->o_valid &= ~OBD_MD_FLHANDLE;
2750
2751         /* OFD/obdfilter works only via prep/commit */
2752         test_mode = (long)data->ioc_pbuf1;
2753         if (!ed->ed_next && test_mode != 3) {
2754                 test_mode = 3;
2755                 data->ioc_plen1 = data->ioc_count;
2756         }
2757
2758         if (test_mode == 3)
2759                 async = 1;
2760
2761         /* Truncate batch size to maximum */
2762         if (data->ioc_plen1 > PTLRPC_MAX_BRW_SIZE)
2763                 data->ioc_plen1 = PTLRPC_MAX_BRW_SIZE;
2764
2765         switch (test_mode) {
2766         case 1:
2767                 /* fall through */
2768         case 2:
2769                 rc = echo_client_kbrw(ed, rw, oa, eco, data->ioc_offset,
2770                                       data->ioc_count, async);
2771                 break;
2772         case 3:
2773                 rc = echo_client_prep_commit(env, ec->ec_exp, rw, oa, eco,
2774                                              data->ioc_offset, data->ioc_count,
2775                                              data->ioc_plen1, async);
2776                 break;
2777         default:
2778                 rc = -EINVAL;
2779         }
2780
2781         echo_put_object(eco);
2782
2783         RETURN(rc);
2784 }
2785
2786 static int
2787 echo_client_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2788                       void *karg, void __user *uarg)
2789 {
2790 #ifdef HAVE_SERVER_SUPPORT
2791         struct tgt_session_info *tsi;
2792 #endif
2793         struct obd_device      *obd = exp->exp_obd;
2794         struct echo_device     *ed = obd2echo_dev(obd);
2795         struct echo_client_obd *ec = ed->ed_ec;
2796         struct echo_object     *eco;
2797         struct obd_ioctl_data  *data = karg;
2798         struct lu_env          *env;
2799         unsigned long           env_tags = 0;
2800         __u16                   refcheck;
2801         struct obdo            *oa;
2802         struct lu_fid           fid;
2803         int                     rw = OBD_BRW_READ;
2804         int                     rc = 0;
2805
2806         ENTRY;
2807         oa = &data->ioc_obdo1;
2808         if (!(oa->o_valid & OBD_MD_FLGROUP)) {
2809                 oa->o_valid |= OBD_MD_FLGROUP;
2810                 ostid_set_seq_echo(&oa->o_oi);
2811         }
2812
2813         /* This FID is unpacked just for validation at this point */
2814         rc = ostid_to_fid(&fid, &oa->o_oi, 0);
2815         if (rc < 0)
2816                 RETURN(rc);
2817
2818         env = cl_env_get(&refcheck);
2819         if (IS_ERR(env))
2820                 RETURN(PTR_ERR(env));
2821
2822         lu_env_add(env);
2823
2824 #ifdef HAVE_SERVER_SUPPORT
2825         if (cmd == OBD_IOC_ECHO_MD || cmd == OBD_IOC_ECHO_ALLOC_SEQ)
2826                 env_tags = ECHO_MD_CTX_TAG;
2827         else
2828 #endif
2829                 env_tags = ECHO_DT_CTX_TAG;
2830
2831         rc = lu_env_refill_by_tags(env, env_tags, ECHO_SES_TAG);
2832         if (rc != 0)
2833                 GOTO(out, rc);
2834
2835 #ifdef HAVE_SERVER_SUPPORT
2836         tsi = tgt_ses_info(env);
2837         /* treat as local operation */
2838         tsi->tsi_exp = NULL;
2839         tsi->tsi_jobid = NULL;
2840 #endif
2841
2842         switch (cmd) {
2843         case OBD_IOC_CREATE:                    /* may create echo object */
2844                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2845                         GOTO(out, rc = -EPERM);
2846
2847                 rc = echo_create_object(env, ed, oa);
2848                 GOTO(out, rc);
2849
2850 #ifdef HAVE_SERVER_SUPPORT
2851         case OBD_IOC_ECHO_MD: {
2852                 int count;
2853                 int cmd;
2854                 char *dir = NULL;
2855                 int dirlen;
2856                 __u64 id;
2857
2858                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2859                         GOTO(out, rc = -EPERM);
2860
2861                 count = data->ioc_count;
2862                 cmd = data->ioc_command;
2863
2864                 id = data->ioc_obdo2.o_oi.oi.oi_id;
2865                 dirlen = data->ioc_plen1;
2866                 OBD_ALLOC(dir, dirlen + 1);
2867                 if (!dir)
2868                         GOTO(out, rc = -ENOMEM);
2869
2870                 if (copy_from_user(dir, data->ioc_pbuf1, dirlen)) {
2871                         OBD_FREE(dir, data->ioc_plen1 + 1);
2872                         GOTO(out, rc = -EFAULT);
2873                 }
2874
2875                 rc = echo_md_handler(ed, cmd, dir, dirlen, id, count, data);
2876                 OBD_FREE(dir, dirlen + 1);
2877                 GOTO(out, rc);
2878         }
2879         case OBD_IOC_ECHO_ALLOC_SEQ: {
2880                 __u64            seq;
2881                 int              max_count;
2882
2883                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2884                         GOTO(out, rc = -EPERM);
2885
2886                 rc = seq_client_get_seq(env, ed->ed_cl_seq, &seq);
2887                 if (rc < 0) {
2888                         CERROR("%s: Can not alloc seq: rc = %d\n",
2889                                obd->obd_name, rc);
2890                         GOTO(out, rc);
2891                 }
2892
2893                 if (copy_to_user(data->ioc_pbuf1, &seq, data->ioc_plen1))
2894                         return -EFAULT;
2895
2896                 max_count = LUSTRE_METADATA_SEQ_MAX_WIDTH;
2897                 if (copy_to_user(data->ioc_pbuf2, &max_count,
2898                                      data->ioc_plen2))
2899                         return -EFAULT;
2900                 GOTO(out, rc);
2901         }
2902 #endif /* HAVE_SERVER_SUPPORT */
2903         case OBD_IOC_DESTROY:
2904                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2905                         GOTO(out, rc = -EPERM);
2906
2907                 rc = echo_get_object(&eco, ed, oa);
2908                 if (rc == 0) {
2909                         rc = obd_destroy(env, ec->ec_exp, oa);
2910                         if (rc == 0)
2911                                 eco->eo_deleted = 1;
2912                         echo_put_object(eco);
2913                 }
2914                 GOTO(out, rc);
2915
2916         case OBD_IOC_GETATTR:
2917                 rc = echo_get_object(&eco, ed, oa);
2918                 if (rc == 0) {
2919                         rc = obd_getattr(env, ec->ec_exp, oa);
2920                         echo_put_object(eco);
2921                 }
2922                 GOTO(out, rc);
2923
2924         case OBD_IOC_SETATTR:
2925                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2926                         GOTO(out, rc = -EPERM);
2927
2928                 rc = echo_get_object(&eco, ed, oa);
2929                 if (rc == 0) {
2930                         rc = obd_setattr(env, ec->ec_exp, oa);
2931                         echo_put_object(eco);
2932                 }
2933                 GOTO(out, rc);
2934
2935         case OBD_IOC_BRW_WRITE:
2936                 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2937                         GOTO(out, rc = -EPERM);
2938
2939                 rw = OBD_BRW_WRITE;
2940                 /* fall through */
2941         case OBD_IOC_BRW_READ:
2942                 rc = echo_client_brw_ioctl(env, rw, exp, data);
2943                 GOTO(out, rc);
2944
2945         default:
2946                 CERROR("echo_ioctl(): unrecognised ioctl %#x\n", cmd);
2947                 GOTO(out, rc = -ENOTTY);
2948         }
2949
2950         EXIT;
2951 out:
2952         lu_env_remove(env);
2953         cl_env_put(env, &refcheck);
2954
2955         return rc;
2956 }
2957
2958 static int echo_client_setup(const struct lu_env *env,
2959                              struct obd_device *obd, struct lustre_cfg *lcfg)
2960 {
2961         struct echo_client_obd *ec = &obd->u.echo_client;
2962         struct obd_device *tgt;
2963         struct obd_uuid echo_uuid = { "ECHO_UUID" };
2964         struct obd_connect_data *ocd = NULL;
2965         int rc;
2966
2967         ENTRY;
2968         if (lcfg->lcfg_bufcount < 2 || LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
2969                 CERROR("requires a TARGET OBD name\n");
2970                 RETURN(-EINVAL);
2971         }
2972
2973         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
2974         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
2975                 CERROR("device not attached or not set up (%s)\n",
2976                        lustre_cfg_string(lcfg, 1));
2977                 RETURN(-EINVAL);
2978         }
2979
2980         spin_lock_init(&ec->ec_lock);
2981         INIT_LIST_HEAD(&ec->ec_objects);
2982         INIT_LIST_HEAD(&ec->ec_locks);
2983         ec->ec_unique = 0;
2984
2985         lu_context_tags_update(ECHO_DT_CTX_TAG);
2986         lu_session_tags_update(ECHO_SES_TAG);
2987
2988         if (!strcmp(tgt->obd_type->typ_name, LUSTRE_MDT_NAME)) {
2989 #ifdef HAVE_SERVER_SUPPORT
2990                 lu_context_tags_update(ECHO_MD_CTX_TAG);
2991 #else
2992                 CERROR(
2993                        "Local operations are NOT supported on client side. Only remote operations are supported. Metadata client must be run on server side.\n");
2994 #endif
2995                 RETURN(0);
2996         }
2997
2998         OBD_ALLOC(ocd, sizeof(*ocd));
2999         if (!ocd) {
3000                 CERROR("Can't alloc ocd connecting to %s\n",
3001                        lustre_cfg_string(lcfg, 1));
3002                 return -ENOMEM;
3003         }
3004
3005         ocd->ocd_connect_flags = OBD_CONNECT_VERSION | OBD_CONNECT_REQPORTAL |
3006                                  OBD_CONNECT_BRW_SIZE |
3007                                  OBD_CONNECT_GRANT | OBD_CONNECT_FULL20 |
3008                                  OBD_CONNECT_64BITHASH | OBD_CONNECT_LVB_TYPE |
3009                                  OBD_CONNECT_FID;
3010         ocd->ocd_brw_size = DT_MAX_BRW_SIZE;
3011         ocd->ocd_version = LUSTRE_VERSION_CODE;
3012         ocd->ocd_group = FID_SEQ_ECHO;
3013
3014         rc = obd_connect(env, &ec->ec_exp, tgt, &echo_uuid, ocd, NULL);
3015         if (rc == 0) {
3016                 /* Turn off pinger because it connects to tgt obd directly. */
3017                 spin_lock(&tgt->obd_dev_lock);
3018                 list_del_init(&ec->ec_exp->exp_obd_chain_timed);
3019                 spin_unlock(&tgt->obd_dev_lock);
3020         }
3021
3022         OBD_FREE(ocd, sizeof(*ocd));
3023
3024         if (rc != 0) {
3025                 CERROR("fail to connect to device %s\n",
3026                        lustre_cfg_string(lcfg, 1));
3027                 return rc;
3028         }
3029
3030         RETURN(rc);
3031 }
3032
3033 static int echo_client_cleanup(struct obd_device *obd)
3034 {
3035         struct echo_device *ed = obd2echo_dev(obd);
3036         struct echo_client_obd *ec = &obd->u.echo_client;
3037         int rc;
3038
3039         ENTRY;
3040         /*Do nothing for Metadata echo client*/
3041         if (!ed)
3042                 RETURN(0);
3043
3044         lu_session_tags_clear(ECHO_SES_TAG & ~LCT_SESSION);
3045         lu_context_tags_clear(ECHO_DT_CTX_TAG);
3046         if (ed->ed_next_ismd) {
3047 #ifdef HAVE_SERVER_SUPPORT
3048                 lu_context_tags_clear(ECHO_MD_CTX_TAG);
3049 #else
3050                 CERROR(
3051                        "This is client-side only module, does not support metadata echo client.\n");
3052 #endif
3053                 RETURN(0);
3054         }
3055
3056         if (!list_empty(&obd->obd_exports)) {
3057                 CERROR("still has clients!\n");
3058                 RETURN(-EBUSY);
3059         }
3060
3061         LASSERT(refcount_read(&ec->ec_exp->exp_handle.h_ref) > 0);
3062         rc = obd_disconnect(ec->ec_exp);
3063         if (rc != 0)
3064                 CERROR("fail to disconnect device: %d\n", rc);
3065
3066         RETURN(rc);
3067 }
3068
3069 static int echo_client_connect(const struct lu_env *env,
3070                                struct obd_export **exp,
3071                                struct obd_device *src, struct obd_uuid *cluuid,
3072                                struct obd_connect_data *data, void *localdata)
3073 {
3074         int rc;
3075         struct lustre_handle conn = { 0 };
3076
3077         ENTRY;
3078         rc = class_connect(&conn, src, cluuid);
3079         if (rc == 0)
3080                 *exp = class_conn2export(&conn);
3081
3082         RETURN(rc);
3083 }
3084
3085 static int echo_client_disconnect(struct obd_export *exp)
3086 {
3087         int rc;
3088
3089         ENTRY;
3090         if (!exp)
3091                 GOTO(out, rc = -EINVAL);
3092
3093         rc = class_disconnect(exp);
3094         GOTO(out, rc);
3095 out:
3096         return rc;
3097 }
3098
3099 static const struct obd_ops echo_client_obd_ops = {
3100         .o_owner       = THIS_MODULE,
3101         .o_iocontrol   = echo_client_iocontrol,
3102         .o_connect     = echo_client_connect,
3103         .o_disconnect  = echo_client_disconnect
3104 };
3105
3106 static int __init obdecho_init(void)
3107 {
3108         int rc;
3109
3110         ENTRY;
3111         LCONSOLE_INFO("Echo OBD driver; http://www.lustre.org/\n");
3112
3113         LASSERT(PAGE_SIZE % OBD_ECHO_BLOCK_SIZE == 0);
3114
3115 # ifdef HAVE_SERVER_SUPPORT
3116         rc = echo_persistent_pages_init();
3117         if (rc != 0)
3118                 goto failed_0;
3119
3120         rc = class_register_type(&echo_obd_ops, NULL, true, NULL,
3121                                  LUSTRE_ECHO_NAME, &echo_srv_type);
3122         if (rc != 0)
3123                 goto failed_1;
3124 # endif
3125
3126         rc = lu_kmem_init(echo_caches);
3127         if (rc == 0) {
3128                 rc = class_register_type(&echo_client_obd_ops, NULL, false,
3129                                          NULL, LUSTRE_ECHO_CLIENT_NAME,
3130                                          &echo_device_type);
3131                 if (rc)
3132                         lu_kmem_fini(echo_caches);
3133         }
3134
3135 # ifdef HAVE_SERVER_SUPPORT
3136         if (rc == 0)
3137                 RETURN(0);
3138
3139         class_unregister_type(LUSTRE_ECHO_NAME);
3140 failed_1:
3141         echo_persistent_pages_fini();
3142 failed_0:
3143 # endif
3144         RETURN(rc);
3145 }
3146
3147 static void __exit obdecho_exit(void)
3148 {
3149         class_unregister_type(LUSTRE_ECHO_CLIENT_NAME);
3150         lu_kmem_fini(echo_caches);
3151
3152 #ifdef HAVE_SERVER_SUPPORT
3153         class_unregister_type(LUSTRE_ECHO_NAME);
3154         echo_persistent_pages_fini();
3155 #endif
3156 }
3157
3158 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
3159 MODULE_DESCRIPTION("Lustre Echo Client test driver");
3160 MODULE_VERSION(LUSTRE_VERSION_STRING);
3161 MODULE_LICENSE("GPL");
3162
3163 module_init(obdecho_init);
3164 module_exit(obdecho_exit);
3165
3166 /** @} echo_client */