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