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