lustre/obdecho/echo_client.c

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