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