4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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.sun.com/software/products/lustre/docs/GPLv2.pdf
20 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21 * CA 95054 USA or visit www.sun.com if you need additional information or
27 * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
28 * Use is subject to license terms.
30 * Copyright (c) 2011, 2014, Intel Corporation.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
37 #define DEBUG_SUBSYSTEM S_ECHO
39 #include <linux/user_namespace.h>
40 #ifdef HAVE_UIDGID_HEADER
41 # include <linux/uidgid.h>
43 #include <libcfs/libcfs.h>
46 #include <obd_support.h>
47 #include <obd_class.h>
48 #include <lustre_debug.h>
49 #include <lprocfs_status.h>
50 #include <cl_object.h>
51 #include <lustre_fid.h>
52 #include <lustre_acl.h>
53 #include <lustre_ioctl.h>
54 #include <lustre_net.h>
55 #ifdef HAVE_SERVER_SUPPORT
56 # include <md_object.h>
57 #endif /* HAVE_SERVER_SUPPORT */
59 #include "echo_internal.h"
61 /** \defgroup echo_client Echo Client
66 struct cl_device ed_cl;
67 struct echo_client_obd *ed_ec;
69 struct cl_site ed_site_myself;
70 struct lu_site *ed_site;
71 struct lu_device *ed_next;
73 struct lu_client_seq *ed_cl_seq;
74 #ifdef HAVE_SERVER_SUPPORT
75 struct local_oid_storage *ed_los;
76 struct lu_fid ed_root_fid;
77 #endif /* HAVE_SERVER_SUPPORT */
81 struct cl_object eo_cl;
82 struct cl_object_header eo_hdr;
83 struct echo_device *eo_dev;
84 struct list_head eo_obj_chain;
85 struct lov_oinfo *eo_oinfo;
90 struct echo_object_conf {
91 struct cl_object_conf eoc_cl;
92 struct lov_oinfo **eoc_oinfo;
96 struct cl_page_slice ep_cl;
101 struct cl_lock_slice el_cl;
102 struct list_head el_chain;
103 struct echo_object *el_object;
105 atomic_t el_refcount;
108 #ifdef HAVE_SERVER_SUPPORT
109 static const char echo_md_root_dir_name[] = "ROOT_ECHO";
112 * In order to use the values of members in struct mdd_device,
113 * we define an alias structure here.
115 struct echo_md_device {
116 struct md_device emd_md_dev;
117 struct obd_export *emd_child_exp;
118 struct dt_device *emd_child;
119 struct dt_device *emd_bottom;
120 struct lu_fid emd_root_fid;
121 struct lu_fid emd_local_root_fid;
123 #endif /* HAVE_SERVER_SUPPORT */
125 static int echo_client_setup(const struct lu_env *env,
126 struct obd_device *obddev,
127 struct lustre_cfg *lcfg);
128 static int echo_client_cleanup(struct obd_device *obddev);
131 /** \defgroup echo_helpers Helper functions
134 static inline struct echo_device *cl2echo_dev(const struct cl_device *dev)
136 return container_of0(dev, struct echo_device, ed_cl);
139 static inline struct cl_device *echo_dev2cl(struct echo_device *d)
144 static inline struct echo_device *obd2echo_dev(const struct obd_device *obd)
146 return cl2echo_dev(lu2cl_dev(obd->obd_lu_dev));
149 static inline struct cl_object *echo_obj2cl(struct echo_object *eco)
154 static inline struct echo_object *cl2echo_obj(const struct cl_object *o)
156 return container_of(o, struct echo_object, eo_cl);
159 static inline struct echo_page *cl2echo_page(const struct cl_page_slice *s)
161 return container_of(s, struct echo_page, ep_cl);
164 static inline struct echo_lock *cl2echo_lock(const struct cl_lock_slice *s)
166 return container_of(s, struct echo_lock, el_cl);
169 static inline struct cl_lock *echo_lock2cl(const struct echo_lock *ecl)
171 return ecl->el_cl.cls_lock;
174 static struct lu_context_key echo_thread_key;
175 static inline struct echo_thread_info *echo_env_info(const struct lu_env *env)
177 struct echo_thread_info *info;
178 info = lu_context_key_get(&env->le_ctx, &echo_thread_key);
179 LASSERT(info != NULL);
184 struct echo_object_conf *cl2echo_conf(const struct cl_object_conf *c)
186 return container_of(c, struct echo_object_conf, eoc_cl);
189 #ifdef HAVE_SERVER_SUPPORT
190 static inline struct echo_md_device *lu2emd_dev(struct lu_device *d)
192 return container_of0(d, struct echo_md_device, emd_md_dev.md_lu_dev);
195 static inline struct lu_device *emd2lu_dev(struct echo_md_device *d)
197 return &d->emd_md_dev.md_lu_dev;
200 static inline struct seq_server_site *echo_md_seq_site(struct echo_md_device *d)
202 return emd2lu_dev(d)->ld_site->ld_seq_site;
205 static inline struct obd_device *emd2obd_dev(struct echo_md_device *d)
207 return d->emd_md_dev.md_lu_dev.ld_obd;
209 #endif /* HAVE_SERVER_SUPPORT */
211 /** @} echo_helpers */
213 static int cl_echo_object_put(struct echo_object *eco);
214 static int cl_echo_object_brw(struct echo_object *eco, int rw, u64 offset,
215 struct page **pages, int npages, int async);
217 struct echo_thread_info {
218 struct echo_object_conf eti_conf;
219 struct lustre_md eti_md;
221 struct cl_2queue eti_queue;
223 struct cl_lock eti_lock;
224 struct lu_fid eti_fid;
225 struct lu_fid eti_fid2;
226 #ifdef HAVE_SERVER_SUPPORT
227 struct md_op_spec eti_spec;
228 struct lov_mds_md_v3 eti_lmm;
229 struct lov_user_md_v3 eti_lum;
230 struct md_attr eti_ma;
231 struct lu_name eti_lname;
232 /* per-thread values, can be re-used */
236 struct lu_buf eti_buf;
237 char eti_xattr_buf[LUSTRE_POSIX_ACL_MAX_SIZE];
241 /* No session used right now */
242 struct echo_session_info {
246 static struct kmem_cache *echo_lock_kmem;
247 static struct kmem_cache *echo_object_kmem;
248 static struct kmem_cache *echo_thread_kmem;
249 static struct kmem_cache *echo_session_kmem;
250 /* static struct kmem_cache *echo_req_kmem; */
252 static struct lu_kmem_descr echo_caches[] = {
254 .ckd_cache = &echo_lock_kmem,
255 .ckd_name = "echo_lock_kmem",
256 .ckd_size = sizeof (struct echo_lock)
259 .ckd_cache = &echo_object_kmem,
260 .ckd_name = "echo_object_kmem",
261 .ckd_size = sizeof (struct echo_object)
264 .ckd_cache = &echo_thread_kmem,
265 .ckd_name = "echo_thread_kmem",
266 .ckd_size = sizeof (struct echo_thread_info)
269 .ckd_cache = &echo_session_kmem,
270 .ckd_name = "echo_session_kmem",
271 .ckd_size = sizeof (struct echo_session_info)
278 /** \defgroup echo_page Page operations
280 * Echo page operations.
284 static int echo_page_own(const struct lu_env *env,
285 const struct cl_page_slice *slice,
286 struct cl_io *io, int nonblock)
288 struct echo_page *ep = cl2echo_page(slice);
291 mutex_lock(&ep->ep_lock);
292 else if (!mutex_trylock(&ep->ep_lock))
297 static void echo_page_disown(const struct lu_env *env,
298 const struct cl_page_slice *slice,
301 struct echo_page *ep = cl2echo_page(slice);
303 LASSERT(mutex_is_locked(&ep->ep_lock));
304 mutex_unlock(&ep->ep_lock);
307 static void echo_page_discard(const struct lu_env *env,
308 const struct cl_page_slice *slice,
309 struct cl_io *unused)
311 cl_page_delete(env, slice->cpl_page);
314 static int echo_page_is_vmlocked(const struct lu_env *env,
315 const struct cl_page_slice *slice)
317 if (mutex_is_locked(&cl2echo_page(slice)->ep_lock))
322 static void echo_page_completion(const struct lu_env *env,
323 const struct cl_page_slice *slice,
326 LASSERT(slice->cpl_page->cp_sync_io != NULL);
329 static void echo_page_fini(const struct lu_env *env,
330 struct cl_page_slice *slice)
332 struct echo_object *eco = cl2echo_obj(slice->cpl_obj);
335 atomic_dec(&eco->eo_npages);
336 page_cache_release(slice->cpl_page->cp_vmpage);
340 static int echo_page_prep(const struct lu_env *env,
341 const struct cl_page_slice *slice,
342 struct cl_io *unused)
347 static int echo_page_print(const struct lu_env *env,
348 const struct cl_page_slice *slice,
349 void *cookie, lu_printer_t printer)
351 struct echo_page *ep = cl2echo_page(slice);
353 (*printer)(env, cookie, LUSTRE_ECHO_CLIENT_NAME"-page@%p %d vm@%p\n",
354 ep, mutex_is_locked(&ep->ep_lock),
355 slice->cpl_page->cp_vmpage);
359 static const struct cl_page_operations echo_page_ops = {
360 .cpo_own = echo_page_own,
361 .cpo_disown = echo_page_disown,
362 .cpo_discard = echo_page_discard,
363 .cpo_fini = echo_page_fini,
364 .cpo_print = echo_page_print,
365 .cpo_is_vmlocked = echo_page_is_vmlocked,
368 .cpo_prep = echo_page_prep,
369 .cpo_completion = echo_page_completion,
372 .cpo_prep = echo_page_prep,
373 .cpo_completion = echo_page_completion,
379 /** \defgroup echo_lock Locking
381 * echo lock operations
385 static void echo_lock_fini(const struct lu_env *env,
386 struct cl_lock_slice *slice)
388 struct echo_lock *ecl = cl2echo_lock(slice);
390 LASSERT(list_empty(&ecl->el_chain));
391 OBD_SLAB_FREE_PTR(ecl, echo_lock_kmem);
394 static struct cl_lock_operations echo_lock_ops = {
395 .clo_fini = echo_lock_fini,
400 /** \defgroup echo_cl_ops cl_object operations
402 * operations for cl_object
406 static int echo_page_init(const struct lu_env *env, struct cl_object *obj,
407 struct cl_page *page, pgoff_t index)
409 struct echo_page *ep = cl_object_page_slice(obj, page);
410 struct echo_object *eco = cl2echo_obj(obj);
413 page_cache_get(page->cp_vmpage);
414 mutex_init(&ep->ep_lock);
415 cl_page_slice_add(page, &ep->ep_cl, obj, index, &echo_page_ops);
416 atomic_inc(&eco->eo_npages);
420 static int echo_io_init(const struct lu_env *env, struct cl_object *obj,
426 static int echo_lock_init(const struct lu_env *env,
427 struct cl_object *obj, struct cl_lock *lock,
428 const struct cl_io *unused)
430 struct echo_lock *el;
433 OBD_SLAB_ALLOC_PTR_GFP(el, echo_lock_kmem, GFP_NOFS);
435 cl_lock_slice_add(lock, &el->el_cl, obj, &echo_lock_ops);
436 el->el_object = cl2echo_obj(obj);
437 INIT_LIST_HEAD(&el->el_chain);
438 atomic_set(&el->el_refcount, 0);
440 RETURN(el == NULL ? -ENOMEM : 0);
443 static int echo_conf_set(const struct lu_env *env, struct cl_object *obj,
444 const struct cl_object_conf *conf)
449 static const struct cl_object_operations echo_cl_obj_ops = {
450 .coo_page_init = echo_page_init,
451 .coo_lock_init = echo_lock_init,
452 .coo_io_init = echo_io_init,
453 .coo_conf_set = echo_conf_set
455 /** @} echo_cl_ops */
457 /** \defgroup echo_lu_ops lu_object operations
459 * operations for echo lu object.
463 static int echo_object_init(const struct lu_env *env, struct lu_object *obj,
464 const struct lu_object_conf *conf)
466 struct echo_device *ed = cl2echo_dev(lu2cl_dev(obj->lo_dev));
467 struct echo_client_obd *ec = ed->ed_ec;
468 struct echo_object *eco = cl2echo_obj(lu2cl(obj));
472 struct lu_object *below;
473 struct lu_device *under;
476 below = under->ld_ops->ldo_object_alloc(env, obj->lo_header,
480 lu_object_add(obj, below);
483 if (!ed->ed_next_ismd) {
484 const struct cl_object_conf *cconf = lu2cl_conf(conf);
485 struct echo_object_conf *econf = cl2echo_conf(cconf);
487 LASSERT(econf->eoc_oinfo != NULL);
489 /* Transfer the oinfo pointer to eco that it won't be
491 eco->eo_oinfo = *econf->eoc_oinfo;
492 *econf->eoc_oinfo = NULL;
494 eco->eo_oinfo = NULL;
498 atomic_set(&eco->eo_npages, 0);
499 cl_object_page_init(lu2cl(obj), sizeof(struct echo_page));
501 spin_lock(&ec->ec_lock);
502 list_add_tail(&eco->eo_obj_chain, &ec->ec_objects);
503 spin_unlock(&ec->ec_lock);
508 static void echo_object_free(const struct lu_env *env, struct lu_object *obj)
510 struct echo_object *eco = cl2echo_obj(lu2cl(obj));
511 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
514 LASSERT(atomic_read(&eco->eo_npages) == 0);
516 spin_lock(&ec->ec_lock);
517 list_del_init(&eco->eo_obj_chain);
518 spin_unlock(&ec->ec_lock);
521 lu_object_header_fini(obj->lo_header);
523 if (eco->eo_oinfo != NULL)
524 OBD_FREE_PTR(eco->eo_oinfo);
526 OBD_SLAB_FREE_PTR(eco, echo_object_kmem);
530 static int echo_object_print(const struct lu_env *env, void *cookie,
531 lu_printer_t p, const struct lu_object *o)
533 struct echo_object *obj = cl2echo_obj(lu2cl(o));
535 return (*p)(env, cookie, "echoclient-object@%p", obj);
538 static const struct lu_object_operations echo_lu_obj_ops = {
539 .loo_object_init = echo_object_init,
540 .loo_object_delete = NULL,
541 .loo_object_release = NULL,
542 .loo_object_free = echo_object_free,
543 .loo_object_print = echo_object_print,
544 .loo_object_invariant = NULL
546 /** @} echo_lu_ops */
548 /** \defgroup echo_lu_dev_ops lu_device operations
550 * Operations for echo lu device.
554 static struct lu_object *echo_object_alloc(const struct lu_env *env,
555 const struct lu_object_header *hdr,
556 struct lu_device *dev)
558 struct echo_object *eco;
559 struct lu_object *obj = NULL;
562 /* we're the top dev. */
563 LASSERT(hdr == NULL);
564 OBD_SLAB_ALLOC_PTR_GFP(eco, echo_object_kmem, GFP_NOFS);
566 struct cl_object_header *hdr = &eco->eo_hdr;
568 obj = &echo_obj2cl(eco)->co_lu;
569 cl_object_header_init(hdr);
570 hdr->coh_page_bufsize = cfs_size_round(sizeof(struct cl_page));
572 lu_object_init(obj, &hdr->coh_lu, dev);
573 lu_object_add_top(&hdr->coh_lu, obj);
575 eco->eo_cl.co_ops = &echo_cl_obj_ops;
576 obj->lo_ops = &echo_lu_obj_ops;
581 static struct lu_device_operations echo_device_lu_ops = {
582 .ldo_object_alloc = echo_object_alloc,
585 /** @} echo_lu_dev_ops */
587 static struct cl_device_operations echo_device_cl_ops = {
590 /** \defgroup echo_init Setup and teardown
592 * Init and fini functions for echo client.
596 static int echo_site_init(const struct lu_env *env, struct echo_device *ed)
598 struct cl_site *site = &ed->ed_site_myself;
601 /* initialize site */
602 rc = cl_site_init(site, &ed->ed_cl);
604 CERROR("Cannot initialize site for echo client(%d)\n", rc);
608 rc = lu_site_init_finish(&site->cs_lu);
614 ed->ed_site = &site->cs_lu;
618 static void echo_site_fini(const struct lu_env *env, struct echo_device *ed)
621 if (!ed->ed_next_ismd)
622 lu_site_fini(ed->ed_site);
627 static void *echo_thread_key_init(const struct lu_context *ctx,
628 struct lu_context_key *key)
630 struct echo_thread_info *info;
632 OBD_SLAB_ALLOC_PTR_GFP(info, echo_thread_kmem, GFP_NOFS);
634 info = ERR_PTR(-ENOMEM);
638 static void echo_thread_key_fini(const struct lu_context *ctx,
639 struct lu_context_key *key, void *data)
641 struct echo_thread_info *info = data;
642 OBD_SLAB_FREE_PTR(info, echo_thread_kmem);
645 static void echo_thread_key_exit(const struct lu_context *ctx,
646 struct lu_context_key *key, void *data)
650 static struct lu_context_key echo_thread_key = {
651 .lct_tags = LCT_CL_THREAD,
652 .lct_init = echo_thread_key_init,
653 .lct_fini = echo_thread_key_fini,
654 .lct_exit = echo_thread_key_exit
657 static void *echo_session_key_init(const struct lu_context *ctx,
658 struct lu_context_key *key)
660 struct echo_session_info *session;
662 OBD_SLAB_ALLOC_PTR_GFP(session, echo_session_kmem, GFP_NOFS);
664 session = ERR_PTR(-ENOMEM);
668 static void echo_session_key_fini(const struct lu_context *ctx,
669 struct lu_context_key *key, void *data)
671 struct echo_session_info *session = data;
672 OBD_SLAB_FREE_PTR(session, echo_session_kmem);
675 static void echo_session_key_exit(const struct lu_context *ctx,
676 struct lu_context_key *key, void *data)
680 static struct lu_context_key echo_session_key = {
681 .lct_tags = LCT_SESSION,
682 .lct_init = echo_session_key_init,
683 .lct_fini = echo_session_key_fini,
684 .lct_exit = echo_session_key_exit
687 LU_TYPE_INIT_FINI(echo, &echo_thread_key, &echo_session_key);
689 #ifdef HAVE_SERVER_SUPPORT
690 # define ECHO_SEQ_WIDTH 0xffffffff
691 static int echo_fid_init(struct echo_device *ed, char *obd_name,
692 struct seq_server_site *ss)
698 OBD_ALLOC_PTR(ed->ed_cl_seq);
699 if (ed->ed_cl_seq == NULL)
702 OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
704 GOTO(out_free_seq, rc = -ENOMEM);
706 snprintf(prefix, MAX_OBD_NAME + 5, "srv-%s", obd_name);
708 /* Init client side sequence-manager */
709 rc = seq_client_init(ed->ed_cl_seq, NULL,
711 prefix, ss->ss_server_seq);
712 ed->ed_cl_seq->lcs_width = ECHO_SEQ_WIDTH;
713 OBD_FREE(prefix, MAX_OBD_NAME + 5);
715 GOTO(out_free_seq, rc);
720 OBD_FREE_PTR(ed->ed_cl_seq);
721 ed->ed_cl_seq = NULL;
725 static int echo_fid_fini(struct obd_device *obddev)
727 struct echo_device *ed = obd2echo_dev(obddev);
730 if (ed->ed_cl_seq != NULL) {
731 seq_client_fini(ed->ed_cl_seq);
732 OBD_FREE_PTR(ed->ed_cl_seq);
733 ed->ed_cl_seq = NULL;
739 static void echo_ed_los_fini(const struct lu_env *env, struct echo_device *ed)
743 if (ed != NULL && ed->ed_next_ismd && ed->ed_los != NULL) {
744 local_oid_storage_fini(env, ed->ed_los);
750 echo_md_local_file_create(const struct lu_env *env, struct echo_md_device *emd,
751 struct local_oid_storage *los,
752 const struct lu_fid *pfid, const char *name,
753 __u32 mode, struct lu_fid *fid)
755 struct dt_object *parent = NULL;
756 struct dt_object *dto = NULL;
760 LASSERT(!fid_is_zero(pfid));
761 parent = dt_locate(env, emd->emd_bottom, pfid);
762 if (unlikely(IS_ERR(parent)))
763 RETURN(PTR_ERR(parent));
765 /* create local file with @fid */
766 dto = local_file_find_or_create_with_fid(env, emd->emd_bottom, fid,
769 GOTO(out_put, rc = PTR_ERR(dto));
771 *fid = *lu_object_fid(&dto->do_lu);
772 /* since stack is not fully set up the local_storage uses own stack
773 * and we should drop its object from cache */
774 lu_object_put_nocache(env, &dto->do_lu);
778 lu_object_put(env, &parent->do_lu);
783 echo_md_root_get(const struct lu_env *env, struct echo_md_device *emd,
784 struct echo_device *ed)
790 /* Setup local dirs */
791 fid.f_seq = FID_SEQ_LOCAL_NAME;
794 rc = local_oid_storage_init(env, emd->emd_bottom, &fid, &ed->ed_los);
798 lu_echo_root_fid(&fid);
799 if (echo_md_seq_site(emd)->ss_node_id == 0) {
800 rc = echo_md_local_file_create(env, emd, ed->ed_los,
801 &emd->emd_local_root_fid,
802 echo_md_root_dir_name, S_IFDIR |
803 S_IRUGO | S_IWUSR | S_IXUGO,
806 CERROR("%s: create md echo root fid failed: rc = %d\n",
807 emd2obd_dev(emd)->obd_name, rc);
811 ed->ed_root_fid = fid;
815 echo_ed_los_fini(env, ed);
819 #endif /* HAVE_SERVER_SUPPORT */
821 static struct lu_device *echo_device_alloc(const struct lu_env *env,
822 struct lu_device_type *t,
823 struct lustre_cfg *cfg)
825 struct lu_device *next;
826 struct echo_device *ed;
827 struct cl_device *cd;
828 struct obd_device *obd = NULL; /* to keep compiler happy */
829 struct obd_device *tgt;
830 const char *tgt_type_name;
837 GOTO(out, rc = -ENOMEM);
841 rc = cl_device_init(cd, t);
845 cd->cd_lu_dev.ld_ops = &echo_device_lu_ops;
846 cd->cd_ops = &echo_device_cl_ops;
849 obd = class_name2obd(lustre_cfg_string(cfg, 0));
850 LASSERT(obd != NULL);
851 LASSERT(env != NULL);
853 tgt = class_name2obd(lustre_cfg_string(cfg, 1));
855 CERROR("Can not find tgt device %s\n",
856 lustre_cfg_string(cfg, 1));
857 GOTO(out, rc = -ENODEV);
860 next = tgt->obd_lu_dev;
862 if (strcmp(tgt->obd_type->typ_name, LUSTRE_MDT_NAME) == 0) {
863 ed->ed_next_ismd = 1;
864 } else if (strcmp(tgt->obd_type->typ_name, LUSTRE_OST_NAME) == 0 ||
865 strcmp(tgt->obd_type->typ_name, LUSTRE_OSC_NAME) == 0) {
866 ed->ed_next_ismd = 0;
867 rc = echo_site_init(env, ed);
871 GOTO(out, rc = -EINVAL);
876 rc = echo_client_setup(env, obd, cfg);
880 ed->ed_ec = &obd->u.echo_client;
883 if (ed->ed_next_ismd) {
884 #ifdef HAVE_SERVER_SUPPORT
885 /* Suppose to connect to some Metadata layer */
886 struct lu_site *ls = NULL;
887 struct lu_device *ld = NULL;
888 struct md_device *md = NULL;
889 struct echo_md_device *emd = NULL;
893 CERROR("%s is not lu device type!\n",
894 lustre_cfg_string(cfg, 1));
895 GOTO(out, rc = -EINVAL);
898 tgt_type_name = lustre_cfg_string(cfg, 2);
899 if (!tgt_type_name) {
900 CERROR("%s no type name for echo %s setup\n",
901 lustre_cfg_string(cfg, 1),
902 tgt->obd_type->typ_name);
903 GOTO(out, rc = -EINVAL);
908 spin_lock(&ls->ls_ld_lock);
909 list_for_each_entry(ld, &ls->ls_ld_linkage, ld_linkage) {
910 if (strcmp(ld->ld_type->ldt_name, tgt_type_name) == 0) {
915 spin_unlock(&ls->ls_ld_lock);
918 CERROR("%s is not lu device type!\n",
919 lustre_cfg_string(cfg, 1));
920 GOTO(out, rc = -EINVAL);
924 /* For MD echo client, it will use the site in MDS stack */
926 ed->ed_cl.cd_lu_dev.ld_site = ls;
927 rc = echo_fid_init(ed, obd->obd_name, lu_site2seq(ls));
929 CERROR("echo fid init error %d\n", rc);
933 md = lu2md_dev(next);
934 emd = lu2emd_dev(&md->md_lu_dev);
935 rc = echo_md_root_get(env, emd, ed);
937 CERROR("%s: get root error: rc = %d\n",
938 emd2obd_dev(emd)->obd_name, rc);
941 #else /* !HAVE_SERVER_SUPPORT */
942 CERROR("Local operations are NOT supported on client side. "
943 "Only remote operations are supported. Metadata client "
944 "must be run on server side.\n");
945 GOTO(out, rc = -EOPNOTSUPP);
946 #endif /* HAVE_SERVER_SUPPORT */
948 /* if echo client is to be stacked upon ost device, the next is
949 * NULL since ost is not a clio device so far */
950 if (next != NULL && !lu_device_is_cl(next))
953 tgt_type_name = tgt->obd_type->typ_name;
955 LASSERT(next != NULL);
956 if (next->ld_site != NULL)
957 GOTO(out, rc = -EBUSY);
959 next->ld_site = ed->ed_site;
960 rc = next->ld_type->ldt_ops->ldto_device_init(env, next,
961 next->ld_type->ldt_name,
966 LASSERT(strcmp(tgt_type_name, LUSTRE_OST_NAME) == 0);
970 RETURN(&cd->cd_lu_dev);
975 rc2 = echo_client_cleanup(obd);
977 CERROR("Cleanup obd device %s error(%d)\n",
982 echo_site_fini(env, ed);
984 cl_device_fini(&ed->ed_cl);
994 static int echo_device_init(const struct lu_env *env, struct lu_device *d,
995 const char *name, struct lu_device *next)
1001 static struct lu_device *echo_device_fini(const struct lu_env *env,
1002 struct lu_device *d)
1004 struct echo_device *ed = cl2echo_dev(lu2cl_dev(d));
1005 struct lu_device *next = ed->ed_next;
1007 while (next && !ed->ed_next_ismd)
1008 next = next->ld_type->ldt_ops->ldto_device_fini(env, next);
1012 static void echo_lock_release(const struct lu_env *env,
1013 struct echo_lock *ecl,
1016 struct cl_lock *clk = echo_lock2cl(ecl);
1018 cl_lock_release(env, clk);
1021 static struct lu_device *echo_device_free(const struct lu_env *env,
1022 struct lu_device *d)
1024 struct echo_device *ed = cl2echo_dev(lu2cl_dev(d));
1025 struct echo_client_obd *ec = ed->ed_ec;
1026 struct echo_object *eco;
1027 struct lu_device *next = ed->ed_next;
1029 CDEBUG(D_INFO, "echo device:%p is going to be freed, next = %p\n",
1032 lu_site_purge(env, ed->ed_site, -1);
1034 /* check if there are objects still alive.
1035 * It shouldn't have any object because lu_site_purge would cleanup
1036 * all of cached objects. Anyway, probably the echo device is being
1037 * parallelly accessed.
1039 spin_lock(&ec->ec_lock);
1040 list_for_each_entry(eco, &ec->ec_objects, eo_obj_chain)
1041 eco->eo_deleted = 1;
1042 spin_unlock(&ec->ec_lock);
1045 lu_site_purge(env, ed->ed_site, -1);
1048 "Waiting for the reference of echo object to be dropped\n");
1050 /* Wait for the last reference to be dropped. */
1051 spin_lock(&ec->ec_lock);
1052 while (!list_empty(&ec->ec_objects)) {
1053 spin_unlock(&ec->ec_lock);
1054 CERROR("echo_client still has objects at cleanup time, "
1055 "wait for 1 second\n");
1056 set_current_state(TASK_UNINTERRUPTIBLE);
1057 schedule_timeout(cfs_time_seconds(1));
1058 lu_site_purge(env, ed->ed_site, -1);
1059 spin_lock(&ec->ec_lock);
1061 spin_unlock(&ec->ec_lock);
1063 LASSERT(list_empty(&ec->ec_locks));
1065 CDEBUG(D_INFO, "No object exists, exiting...\n");
1067 echo_client_cleanup(d->ld_obd);
1068 #ifdef HAVE_SERVER_SUPPORT
1069 echo_fid_fini(d->ld_obd);
1070 echo_ed_los_fini(env, ed);
1072 while (next && !ed->ed_next_ismd)
1073 next = next->ld_type->ldt_ops->ldto_device_free(env, next);
1075 LASSERT(ed->ed_site == d->ld_site);
1076 echo_site_fini(env, ed);
1077 cl_device_fini(&ed->ed_cl);
1083 static const struct lu_device_type_operations echo_device_type_ops = {
1084 .ldto_init = echo_type_init,
1085 .ldto_fini = echo_type_fini,
1087 .ldto_start = echo_type_start,
1088 .ldto_stop = echo_type_stop,
1090 .ldto_device_alloc = echo_device_alloc,
1091 .ldto_device_free = echo_device_free,
1092 .ldto_device_init = echo_device_init,
1093 .ldto_device_fini = echo_device_fini
1096 static struct lu_device_type echo_device_type = {
1097 .ldt_tags = LU_DEVICE_CL,
1098 .ldt_name = LUSTRE_ECHO_CLIENT_NAME,
1099 .ldt_ops = &echo_device_type_ops,
1100 .ldt_ctx_tags = LCT_CL_THREAD | LCT_MD_THREAD | LCT_DT_THREAD,
1104 /** \defgroup echo_exports Exported operations
1106 * exporting functions to echo client
1111 /* Interfaces to echo client obd device */
1112 static struct echo_object *
1113 cl_echo_object_find(struct echo_device *d, const struct ost_id *oi)
1116 struct echo_thread_info *info;
1117 struct echo_object_conf *conf;
1118 struct echo_object *eco;
1119 struct cl_object *obj;
1120 struct lov_oinfo *oinfo = NULL;
1126 LASSERTF(ostid_id(oi) != 0, DOSTID"\n", POSTID(oi));
1127 LASSERTF(ostid_seq(oi) == FID_SEQ_ECHO, DOSTID"\n", POSTID(oi));
1129 /* Never return an object if the obd is to be freed. */
1130 if (echo_dev2cl(d)->cd_lu_dev.ld_obd->obd_stopping)
1131 RETURN(ERR_PTR(-ENODEV));
1133 env = cl_env_get(&refcheck);
1135 RETURN((void *)env);
1137 info = echo_env_info(env);
1138 conf = &info->eti_conf;
1140 OBD_ALLOC_PTR(oinfo);
1142 GOTO(out, eco = ERR_PTR(-ENOMEM));
1144 oinfo->loi_oi = *oi;
1145 conf->eoc_cl.u.coc_oinfo = oinfo;
1148 /* If echo_object_init() is successful then ownership of oinfo
1149 * is transferred to the object. */
1150 conf->eoc_oinfo = &oinfo;
1152 fid = &info->eti_fid;
1153 rc = ostid_to_fid(fid, oi, 0);
1155 GOTO(out, eco = ERR_PTR(rc));
1157 /* In the function below, .hs_keycmp resolves to
1158 * lu_obj_hop_keycmp() */
1159 /* coverity[overrun-buffer-val] */
1160 obj = cl_object_find(env, echo_dev2cl(d), fid, &conf->eoc_cl);
1162 GOTO(out, eco = (void*)obj);
1164 eco = cl2echo_obj(obj);
1165 if (eco->eo_deleted) {
1166 cl_object_put(env, obj);
1167 eco = ERR_PTR(-EAGAIN);
1172 OBD_FREE_PTR(oinfo);
1174 cl_env_put(env, &refcheck);
1178 static int cl_echo_object_put(struct echo_object *eco)
1181 struct cl_object *obj = echo_obj2cl(eco);
1185 env = cl_env_get(&refcheck);
1187 RETURN(PTR_ERR(env));
1189 /* an external function to kill an object? */
1190 if (eco->eo_deleted) {
1191 struct lu_object_header *loh = obj->co_lu.lo_header;
1192 LASSERT(&eco->eo_hdr == luh2coh(loh));
1193 set_bit(LU_OBJECT_HEARD_BANSHEE, &loh->loh_flags);
1196 cl_object_put(env, obj);
1197 cl_env_put(env, &refcheck);
1201 static int cl_echo_enqueue0(struct lu_env *env, struct echo_object *eco,
1202 u64 start, u64 end, int mode,
1203 __u64 *cookie , __u32 enqflags)
1206 struct cl_lock *lck;
1207 struct cl_object *obj;
1208 struct cl_lock_descr *descr;
1209 struct echo_thread_info *info;
1213 info = echo_env_info(env);
1215 lck = &info->eti_lock;
1216 obj = echo_obj2cl(eco);
1218 memset(lck, 0, sizeof(*lck));
1219 descr = &lck->cll_descr;
1220 descr->cld_obj = obj;
1221 descr->cld_start = cl_index(obj, start);
1222 descr->cld_end = cl_index(obj, end);
1223 descr->cld_mode = mode == LCK_PW ? CLM_WRITE : CLM_READ;
1224 descr->cld_enq_flags = enqflags;
1227 rc = cl_lock_request(env, io, lck);
1229 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
1230 struct echo_lock *el;
1232 el = cl2echo_lock(cl_lock_at(lck, &echo_device_type));
1233 spin_lock(&ec->ec_lock);
1234 if (list_empty(&el->el_chain)) {
1235 list_add(&el->el_chain, &ec->ec_locks);
1236 el->el_cookie = ++ec->ec_unique;
1238 atomic_inc(&el->el_refcount);
1239 *cookie = el->el_cookie;
1240 spin_unlock(&ec->ec_lock);
1245 static int cl_echo_cancel0(struct lu_env *env, struct echo_device *ed,
1248 struct echo_client_obd *ec = ed->ed_ec;
1249 struct echo_lock *ecl = NULL;
1250 struct list_head *el;
1251 int found = 0, still_used = 0;
1254 LASSERT(ec != NULL);
1255 spin_lock(&ec->ec_lock);
1256 list_for_each(el, &ec->ec_locks) {
1257 ecl = list_entry(el, struct echo_lock, el_chain);
1258 CDEBUG(D_INFO, "ecl: %p, cookie: "LPX64"\n", ecl, ecl->el_cookie);
1259 found = (ecl->el_cookie == cookie);
1261 if (atomic_dec_and_test(&ecl->el_refcount))
1262 list_del_init(&ecl->el_chain);
1268 spin_unlock(&ec->ec_lock);
1273 echo_lock_release(env, ecl, still_used);
1277 static void echo_commit_callback(const struct lu_env *env, struct cl_io *io,
1278 struct cl_page *page)
1280 struct echo_thread_info *info;
1281 struct cl_2queue *queue;
1283 info = echo_env_info(env);
1284 LASSERT(io == &info->eti_io);
1286 queue = &info->eti_queue;
1287 cl_page_list_add(&queue->c2_qout, page);
1290 static int cl_echo_object_brw(struct echo_object *eco, int rw, u64 offset,
1291 struct page **pages, int npages, int async)
1294 struct echo_thread_info *info;
1295 struct cl_object *obj = echo_obj2cl(eco);
1296 struct echo_device *ed = eco->eo_dev;
1297 struct cl_2queue *queue;
1299 struct cl_page *clp;
1300 struct lustre_handle lh = { 0 };
1301 int page_size = cl_page_size(obj);
1307 LASSERT((offset & ~PAGE_MASK) == 0);
1308 LASSERT(ed->ed_next != NULL);
1309 env = cl_env_get(&refcheck);
1311 RETURN(PTR_ERR(env));
1313 info = echo_env_info(env);
1315 queue = &info->eti_queue;
1317 cl_2queue_init(queue);
1319 io->ci_ignore_layout = 1;
1320 rc = cl_io_init(env, io, CIT_MISC, obj);
1326 rc = cl_echo_enqueue0(env, eco, offset,
1327 offset + npages * PAGE_CACHE_SIZE - 1,
1328 rw == READ ? LCK_PR : LCK_PW, &lh.cookie,
1331 GOTO(error_lock, rc);
1333 for (i = 0; i < npages; i++) {
1335 clp = cl_page_find(env, obj, cl_index(obj, offset),
1336 pages[i], CPT_TRANSIENT);
1341 LASSERT(clp->cp_type == CPT_TRANSIENT);
1343 rc = cl_page_own(env, io, clp);
1345 LASSERT(clp->cp_state == CPS_FREEING);
1346 cl_page_put(env, clp);
1350 cl_2queue_add(queue, clp);
1352 /* drop the reference count for cl_page_find, so that the page
1353 * will be freed in cl_2queue_fini. */
1354 cl_page_put(env, clp);
1355 cl_page_clip(env, clp, 0, page_size);
1357 offset += page_size;
1361 enum cl_req_type typ = rw == READ ? CRT_READ : CRT_WRITE;
1363 async = async && (typ == CRT_WRITE);
1365 rc = cl_io_commit_async(env, io, &queue->c2_qin,
1367 echo_commit_callback);
1369 rc = cl_io_submit_sync(env, io, typ, queue, 0);
1370 CDEBUG(D_INFO, "echo_client %s write returns %d\n",
1371 async ? "async" : "sync", rc);
1374 cl_echo_cancel0(env, ed, lh.cookie);
1377 cl_2queue_discard(env, io, queue);
1378 cl_2queue_disown(env, io, queue);
1379 cl_2queue_fini(env, queue);
1380 cl_io_fini(env, io);
1382 cl_env_put(env, &refcheck);
1385 /** @} echo_exports */
1388 static u64 last_object_id;
1390 #ifdef HAVE_SERVER_SUPPORT
1391 static inline void echo_md_build_name(struct lu_name *lname, char *name,
1394 sprintf(name, LPU64, id);
1395 lname->ln_name = name;
1396 lname->ln_namelen = strlen(name);
1399 /* similar to mdt_attr_get_complex */
1400 static int echo_big_lmm_get(const struct lu_env *env, struct md_object *o,
1403 struct echo_thread_info *info = echo_env_info(env);
1408 LASSERT(ma->ma_lmm_size > 0);
1410 rc = mo_xattr_get(env, o, &LU_BUF_NULL, XATTR_NAME_LOV);
1414 /* big_lmm may need to be grown */
1415 if (info->eti_big_lmmsize < rc) {
1416 int size = size_roundup_power2(rc);
1418 if (info->eti_big_lmmsize > 0) {
1419 /* free old buffer */
1420 LASSERT(info->eti_big_lmm);
1421 OBD_FREE_LARGE(info->eti_big_lmm,
1422 info->eti_big_lmmsize);
1423 info->eti_big_lmm = NULL;
1424 info->eti_big_lmmsize = 0;
1427 OBD_ALLOC_LARGE(info->eti_big_lmm, size);
1428 if (info->eti_big_lmm == NULL)
1430 info->eti_big_lmmsize = size;
1432 LASSERT(info->eti_big_lmmsize >= rc);
1434 info->eti_buf.lb_buf = info->eti_big_lmm;
1435 info->eti_buf.lb_len = info->eti_big_lmmsize;
1436 rc = mo_xattr_get(env, o, &info->eti_buf, XATTR_NAME_LOV);
1440 ma->ma_valid |= MA_LOV;
1441 ma->ma_lmm = info->eti_big_lmm;
1442 ma->ma_lmm_size = rc;
1447 static int echo_attr_get_complex(const struct lu_env *env,
1448 struct md_object *next,
1451 struct echo_thread_info *info = echo_env_info(env);
1452 struct lu_buf *buf = &info->eti_buf;
1453 umode_t mode = lu_object_attr(&next->mo_lu);
1454 int need = ma->ma_need;
1461 if (need & MA_INODE) {
1462 ma->ma_need = MA_INODE;
1463 rc = mo_attr_get(env, next, ma);
1466 ma->ma_valid |= MA_INODE;
1469 if (need & MA_LOV) {
1470 if (S_ISREG(mode) || S_ISDIR(mode)) {
1471 LASSERT(ma->ma_lmm_size > 0);
1472 buf->lb_buf = ma->ma_lmm;
1473 buf->lb_len = ma->ma_lmm_size;
1474 rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_LOV);
1476 ma->ma_lmm_size = rc2;
1477 ma->ma_valid |= MA_LOV;
1478 } else if (rc2 == -ENODATA) {
1480 ma->ma_lmm_size = 0;
1481 } else if (rc2 == -ERANGE) {
1482 rc2 = echo_big_lmm_get(env, next, ma);
1484 GOTO(out, rc = rc2);
1486 GOTO(out, rc = rc2);
1491 #ifdef CONFIG_FS_POSIX_ACL
1492 if (need & MA_ACL_DEF && S_ISDIR(mode)) {
1493 buf->lb_buf = ma->ma_acl;
1494 buf->lb_len = ma->ma_acl_size;
1495 rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_ACL_DEFAULT);
1497 ma->ma_acl_size = rc2;
1498 ma->ma_valid |= MA_ACL_DEF;
1499 } else if (rc2 == -ENODATA) {
1501 ma->ma_acl_size = 0;
1503 GOTO(out, rc = rc2);
1509 CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64" ma_lmm=%p\n",
1510 rc, ma->ma_valid, ma->ma_lmm);
1515 echo_md_create_internal(const struct lu_env *env, struct echo_device *ed,
1516 struct md_object *parent, struct lu_fid *fid,
1517 struct lu_name *lname, struct md_op_spec *spec,
1520 struct lu_object *ec_child, *child;
1521 struct lu_device *ld = ed->ed_next;
1522 struct echo_thread_info *info = echo_env_info(env);
1523 struct lu_fid *fid2 = &info->eti_fid2;
1524 struct lu_object_conf conf = { .loc_flags = LOC_F_NEW };
1529 rc = mdo_lookup(env, parent, lname, fid2, spec);
1532 else if (rc != -ENOENT)
1535 ec_child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev,
1537 if (IS_ERR(ec_child)) {
1538 CERROR("Can not find the child "DFID": rc = %ld\n", PFID(fid),
1540 RETURN(PTR_ERR(ec_child));
1543 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1544 if (child == NULL) {
1545 CERROR("Can not locate the child "DFID"\n", PFID(fid));
1546 GOTO(out_put, rc = -EINVAL);
1549 CDEBUG(D_RPCTRACE, "Start creating object "DFID" %s %p\n",
1550 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
1553 * Do not perform lookup sanity check. We know that name does not exist.
1555 spec->sp_cr_lookup = 0;
1556 rc = mdo_create(env, parent, lname, lu2md(child), spec, ma);
1558 CERROR("Can not create child "DFID": rc = %d\n", PFID(fid), rc);
1561 CDEBUG(D_RPCTRACE, "End creating object "DFID" %s %p rc = %d\n",
1562 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent, rc);
1565 lu_object_put(env, ec_child);
1569 static int echo_set_lmm_size(const struct lu_env *env, struct lu_device *ld,
1572 struct echo_thread_info *info = echo_env_info(env);
1574 if (strcmp(ld->ld_type->ldt_name, LUSTRE_MDD_NAME)) {
1575 ma->ma_lmm = (void *)&info->eti_lmm;
1576 ma->ma_lmm_size = sizeof(info->eti_lmm);
1578 LASSERT(info->eti_big_lmmsize);
1579 ma->ma_lmm = info->eti_big_lmm;
1580 ma->ma_lmm_size = info->eti_big_lmmsize;
1586 static int echo_create_md_object(const struct lu_env *env,
1587 struct echo_device *ed,
1588 struct lu_object *ec_parent,
1590 char *name, int namelen,
1591 __u64 id, __u32 mode, int count,
1592 int stripe_count, int stripe_offset)
1594 struct lu_object *parent;
1595 struct echo_thread_info *info = echo_env_info(env);
1596 struct lu_name *lname = &info->eti_lname;
1597 struct md_op_spec *spec = &info->eti_spec;
1598 struct md_attr *ma = &info->eti_ma;
1599 struct lu_device *ld = ed->ed_next;
1605 if (ec_parent == NULL)
1607 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1611 memset(ma, 0, sizeof(*ma));
1612 memset(spec, 0, sizeof(*spec));
1613 if (stripe_count != 0) {
1614 spec->sp_cr_flags |= FMODE_WRITE;
1615 echo_set_lmm_size(env, ld, ma);
1616 if (stripe_count != -1) {
1617 struct lov_user_md_v3 *lum = &info->eti_lum;
1619 lum->lmm_magic = LOV_USER_MAGIC_V3;
1620 lum->lmm_stripe_count = stripe_count;
1621 lum->lmm_stripe_offset = stripe_offset;
1622 lum->lmm_pattern = 0;
1623 spec->u.sp_ea.eadata = lum;
1624 spec->u.sp_ea.eadatalen = sizeof(*lum);
1625 spec->sp_cr_flags |= MDS_OPEN_HAS_EA;
1629 ma->ma_attr.la_mode = mode;
1630 ma->ma_attr.la_valid = LA_CTIME | LA_MODE;
1631 ma->ma_attr.la_ctime = cfs_time_current_64();
1634 lname->ln_name = name;
1635 lname->ln_namelen = namelen;
1636 /* If name is specified, only create one object by name */
1637 rc = echo_md_create_internal(env, ed, lu2md(parent), fid, lname,
1642 /* Create multiple object sequenced by id */
1643 for (i = 0; i < count; i++) {
1644 char *tmp_name = info->eti_name;
1646 echo_md_build_name(lname, tmp_name, id);
1648 rc = echo_md_create_internal(env, ed, lu2md(parent), fid, lname,
1651 CERROR("Can not create child %s: rc = %d\n", tmp_name,
1662 static struct lu_object *echo_md_lookup(const struct lu_env *env,
1663 struct echo_device *ed,
1664 struct md_object *parent,
1665 struct lu_name *lname)
1667 struct echo_thread_info *info = echo_env_info(env);
1668 struct lu_fid *fid = &info->eti_fid;
1669 struct lu_object *child;
1673 CDEBUG(D_INFO, "lookup %s in parent "DFID" %p\n", lname->ln_name,
1675 rc = mdo_lookup(env, parent, lname, fid, NULL);
1677 CERROR("lookup %s: rc = %d\n", lname->ln_name, rc);
1678 RETURN(ERR_PTR(rc));
1681 /* In the function below, .hs_keycmp resolves to
1682 * lu_obj_hop_keycmp() */
1683 /* coverity[overrun-buffer-val] */
1684 child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, fid, NULL);
1689 static int echo_setattr_object(const struct lu_env *env,
1690 struct echo_device *ed,
1691 struct lu_object *ec_parent,
1692 __u64 id, int count)
1694 struct lu_object *parent;
1695 struct echo_thread_info *info = echo_env_info(env);
1696 struct lu_name *lname = &info->eti_lname;
1697 char *name = info->eti_name;
1698 struct lu_device *ld = ed->ed_next;
1699 struct lu_buf *buf = &info->eti_buf;
1705 if (ec_parent == NULL)
1707 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1711 for (i = 0; i < count; i++) {
1712 struct lu_object *ec_child, *child;
1714 echo_md_build_name(lname, name, id);
1716 ec_child = echo_md_lookup(env, ed, lu2md(parent), lname);
1717 if (IS_ERR(ec_child)) {
1718 CERROR("Can't find child %s: rc = %ld\n",
1719 lname->ln_name, PTR_ERR(ec_child));
1720 RETURN(PTR_ERR(ec_child));
1723 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1724 if (child == NULL) {
1725 CERROR("Can not locate the child %s\n", lname->ln_name);
1726 lu_object_put(env, ec_child);
1731 CDEBUG(D_RPCTRACE, "Start setattr object "DFID"\n",
1732 PFID(lu_object_fid(child)));
1734 buf->lb_buf = info->eti_xattr_buf;
1735 buf->lb_len = sizeof(info->eti_xattr_buf);
1737 sprintf(name, "%s.test1", XATTR_USER_PREFIX);
1738 rc = mo_xattr_set(env, lu2md(child), buf, name,
1741 CERROR("Can not setattr child "DFID": rc = %d\n",
1742 PFID(lu_object_fid(child)), rc);
1743 lu_object_put(env, ec_child);
1746 CDEBUG(D_RPCTRACE, "End setattr object "DFID"\n",
1747 PFID(lu_object_fid(child)));
1749 lu_object_put(env, ec_child);
1754 static int echo_getattr_object(const struct lu_env *env,
1755 struct echo_device *ed,
1756 struct lu_object *ec_parent,
1757 __u64 id, int count)
1759 struct lu_object *parent;
1760 struct echo_thread_info *info = echo_env_info(env);
1761 struct lu_name *lname = &info->eti_lname;
1762 char *name = info->eti_name;
1763 struct md_attr *ma = &info->eti_ma;
1764 struct lu_device *ld = ed->ed_next;
1770 if (ec_parent == NULL)
1772 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1776 memset(ma, 0, sizeof(*ma));
1777 ma->ma_need |= MA_INODE | MA_LOV | MA_PFID | MA_HSM | MA_ACL_DEF;
1778 ma->ma_acl = info->eti_xattr_buf;
1779 ma->ma_acl_size = sizeof(info->eti_xattr_buf);
1781 for (i = 0; i < count; i++) {
1782 struct lu_object *ec_child, *child;
1785 echo_md_build_name(lname, name, id);
1786 echo_set_lmm_size(env, ld, ma);
1788 ec_child = echo_md_lookup(env, ed, lu2md(parent), lname);
1789 if (IS_ERR(ec_child)) {
1790 CERROR("Can't find child %s: rc = %ld\n",
1791 lname->ln_name, PTR_ERR(ec_child));
1792 RETURN(PTR_ERR(ec_child));
1795 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1796 if (child == NULL) {
1797 CERROR("Can not locate the child %s\n", lname->ln_name);
1798 lu_object_put(env, ec_child);
1802 CDEBUG(D_RPCTRACE, "Start getattr object "DFID"\n",
1803 PFID(lu_object_fid(child)));
1804 rc = echo_attr_get_complex(env, lu2md(child), ma);
1806 CERROR("Can not getattr child "DFID": rc = %d\n",
1807 PFID(lu_object_fid(child)), rc);
1808 lu_object_put(env, ec_child);
1811 CDEBUG(D_RPCTRACE, "End getattr object "DFID"\n",
1812 PFID(lu_object_fid(child)));
1814 lu_object_put(env, ec_child);
1820 static int echo_lookup_object(const struct lu_env *env,
1821 struct echo_device *ed,
1822 struct lu_object *ec_parent,
1823 __u64 id, int count)
1825 struct lu_object *parent;
1826 struct echo_thread_info *info = echo_env_info(env);
1827 struct lu_name *lname = &info->eti_lname;
1828 char *name = info->eti_name;
1829 struct lu_fid *fid = &info->eti_fid;
1830 struct lu_device *ld = ed->ed_next;
1834 if (ec_parent == NULL)
1836 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1840 /*prepare the requests*/
1841 for (i = 0; i < count; i++) {
1842 echo_md_build_name(lname, name, id);
1844 CDEBUG(D_RPCTRACE, "Start lookup object "DFID" %s %p\n",
1845 PFID(lu_object_fid(parent)), lname->ln_name, parent);
1847 rc = mdo_lookup(env, lu2md(parent), lname, fid, NULL);
1849 CERROR("Can not lookup child %s: rc = %d\n", name, rc);
1852 CDEBUG(D_RPCTRACE, "End lookup object "DFID" %s %p\n",
1853 PFID(lu_object_fid(parent)), lname->ln_name, parent);
1860 static int echo_md_destroy_internal(const struct lu_env *env,
1861 struct echo_device *ed,
1862 struct md_object *parent,
1863 struct lu_name *lname,
1866 struct lu_device *ld = ed->ed_next;
1867 struct lu_object *ec_child;
1868 struct lu_object *child;
1873 ec_child = echo_md_lookup(env, ed, parent, lname);
1874 if (IS_ERR(ec_child)) {
1875 CERROR("Can't find child %s: rc = %ld\n", lname->ln_name,
1877 RETURN(PTR_ERR(ec_child));
1880 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1881 if (child == NULL) {
1882 CERROR("Can not locate the child %s\n", lname->ln_name);
1883 GOTO(out_put, rc = -EINVAL);
1886 if (lu_object_remote(child)) {
1887 CERROR("Can not destroy remote object %s: rc = %d\n",
1888 lname->ln_name, -EPERM);
1889 GOTO(out_put, rc = -EPERM);
1891 CDEBUG(D_RPCTRACE, "Start destroy object "DFID" %s %p\n",
1892 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
1894 rc = mdo_unlink(env, parent, lu2md(child), lname, ma, 0);
1896 CERROR("Can not unlink child %s: rc = %d\n",
1897 lname->ln_name, rc);
1900 CDEBUG(D_RPCTRACE, "End destroy object "DFID" %s %p\n",
1901 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
1903 lu_object_put(env, ec_child);
1907 static int echo_destroy_object(const struct lu_env *env,
1908 struct echo_device *ed,
1909 struct lu_object *ec_parent,
1910 char *name, int namelen,
1911 __u64 id, __u32 mode,
1914 struct echo_thread_info *info = echo_env_info(env);
1915 struct lu_name *lname = &info->eti_lname;
1916 struct md_attr *ma = &info->eti_ma;
1917 struct lu_device *ld = ed->ed_next;
1918 struct lu_object *parent;
1923 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1927 memset(ma, 0, sizeof(*ma));
1928 ma->ma_attr.la_mode = mode;
1929 ma->ma_attr.la_valid = LA_CTIME;
1930 ma->ma_attr.la_ctime = cfs_time_current_64();
1931 ma->ma_need = MA_INODE;
1935 lname->ln_name = name;
1936 lname->ln_namelen = namelen;
1937 rc = echo_md_destroy_internal(env, ed, lu2md(parent), lname,
1942 /*prepare the requests*/
1943 for (i = 0; i < count; i++) {
1944 char *tmp_name = info->eti_name;
1947 echo_md_build_name(lname, tmp_name, id);
1949 rc = echo_md_destroy_internal(env, ed, lu2md(parent), lname,
1952 CERROR("Can not unlink child %s: rc = %d\n", name, rc);
1961 static struct lu_object *echo_resolve_path(const struct lu_env *env,
1962 struct echo_device *ed, char *path,
1965 struct lu_device *ld = ed->ed_next;
1966 struct echo_thread_info *info = echo_env_info(env);
1967 struct lu_fid *fid = &info->eti_fid;
1968 struct lu_name *lname = &info->eti_lname;
1969 struct lu_object *parent = NULL;
1970 struct lu_object *child = NULL;
1974 *fid = ed->ed_root_fid;
1976 /* In the function below, .hs_keycmp resolves to
1977 * lu_obj_hop_keycmp() */
1978 /* coverity[overrun-buffer-val] */
1979 parent = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, fid, NULL);
1980 if (IS_ERR(parent)) {
1981 CERROR("Can not find the parent "DFID": rc = %ld\n",
1982 PFID(fid), PTR_ERR(parent));
1987 struct lu_object *ld_parent;
1990 e = strsep(&path, "/");
1995 if (!path || path[0] == '\0')
2001 lname->ln_namelen = strlen(e);
2003 ld_parent = lu_object_locate(parent->lo_header, ld->ld_type);
2004 if (ld_parent == NULL) {
2005 lu_object_put(env, parent);
2010 child = echo_md_lookup(env, ed, lu2md(ld_parent), lname);
2011 lu_object_put(env, parent);
2012 if (IS_ERR(child)) {
2013 rc = (int)PTR_ERR(child);
2014 CERROR("lookup %s under parent "DFID": rc = %d\n",
2015 lname->ln_name, PFID(lu_object_fid(ld_parent)),
2022 RETURN(ERR_PTR(rc));
2027 static void echo_ucred_init(struct lu_env *env)
2029 struct lu_ucred *ucred = lu_ucred(env);
2031 ucred->uc_valid = UCRED_INVALID;
2033 ucred->uc_suppgids[0] = -1;
2034 ucred->uc_suppgids[1] = -1;
2036 ucred->uc_uid = ucred->uc_o_uid =
2037 from_kuid(&init_user_ns, current_uid());
2038 ucred->uc_gid = ucred->uc_o_gid =
2039 from_kgid(&init_user_ns, current_gid());
2040 ucred->uc_fsuid = ucred->uc_o_fsuid =
2041 from_kuid(&init_user_ns, current_fsuid());
2042 ucred->uc_fsgid = ucred->uc_o_fsgid =
2043 from_kgid(&init_user_ns, current_fsgid());
2044 ucred->uc_cap = cfs_curproc_cap_pack();
2046 /* remove fs privilege for non-root user. */
2047 if (ucred->uc_fsuid)
2048 ucred->uc_cap &= ~CFS_CAP_FS_MASK;
2049 ucred->uc_valid = UCRED_NEW;
2052 static void echo_ucred_fini(struct lu_env *env)
2054 struct lu_ucred *ucred = lu_ucred(env);
2055 ucred->uc_valid = UCRED_INIT;
2058 #define ECHO_MD_CTX_TAG (LCT_REMEMBER | LCT_MD_THREAD)
2059 #define ECHO_MD_SES_TAG (LCT_REMEMBER | LCT_SESSION | LCT_SERVER_SESSION)
2060 static int echo_md_handler(struct echo_device *ed, int command,
2061 char *path, int path_len, __u64 id, int count,
2062 struct obd_ioctl_data *data)
2064 struct echo_thread_info *info;
2065 struct lu_device *ld = ed->ed_next;
2068 struct lu_object *parent;
2070 int namelen = data->ioc_plen2;
2075 CERROR("MD echo client is not being initialized properly\n");
2079 if (strcmp(ld->ld_type->ldt_name, LUSTRE_MDD_NAME)) {
2080 CERROR("Only support MDD layer right now!\n");
2084 env = cl_env_get(&refcheck);
2086 RETURN(PTR_ERR(env));
2088 rc = lu_env_refill_by_tags(env, ECHO_MD_CTX_TAG, ECHO_MD_SES_TAG);
2092 /* init big_lmm buffer */
2093 info = echo_env_info(env);
2094 LASSERT(info->eti_big_lmm == NULL);
2095 OBD_ALLOC_LARGE(info->eti_big_lmm, MIN_MD_SIZE);
2096 if (info->eti_big_lmm == NULL)
2097 GOTO(out_env, rc = -ENOMEM);
2098 info->eti_big_lmmsize = MIN_MD_SIZE;
2100 parent = echo_resolve_path(env, ed, path, path_len);
2101 if (IS_ERR(parent)) {
2102 CERROR("Can not resolve the path %s: rc = %ld\n", path,
2104 GOTO(out_free, rc = PTR_ERR(parent));
2108 OBD_ALLOC(name, namelen + 1);
2110 GOTO(out_put, rc = -ENOMEM);
2111 if (copy_from_user(name, data->ioc_pbuf2, namelen))
2112 GOTO(out_name, rc = -EFAULT);
2115 echo_ucred_init(env);
2118 case ECHO_MD_CREATE:
2119 case ECHO_MD_MKDIR: {
2120 struct echo_thread_info *info = echo_env_info(env);
2121 __u32 mode = data->ioc_obdo2.o_mode;
2122 struct lu_fid *fid = &info->eti_fid;
2123 int stripe_count = (int)data->ioc_obdo2.o_misc;
2124 int stripe_index = (int)data->ioc_obdo2.o_stripe_idx;
2126 rc = ostid_to_fid(fid, &data->ioc_obdo1.o_oi, 0);
2130 /* In the function below, .hs_keycmp resolves to
2131 * lu_obj_hop_keycmp() */
2132 /* coverity[overrun-buffer-val] */
2133 rc = echo_create_md_object(env, ed, parent, fid, name, namelen,
2134 id, mode, count, stripe_count,
2138 case ECHO_MD_DESTROY:
2139 case ECHO_MD_RMDIR: {
2140 __u32 mode = data->ioc_obdo2.o_mode;
2142 rc = echo_destroy_object(env, ed, parent, name, namelen,
2146 case ECHO_MD_LOOKUP:
2147 rc = echo_lookup_object(env, ed, parent, id, count);
2149 case ECHO_MD_GETATTR:
2150 rc = echo_getattr_object(env, ed, parent, id, count);
2152 case ECHO_MD_SETATTR:
2153 rc = echo_setattr_object(env, ed, parent, id, count);
2156 CERROR("unknown command %d\n", command);
2160 echo_ucred_fini(env);
2164 OBD_FREE(name, namelen + 1);
2166 lu_object_put(env, parent);
2168 LASSERT(info->eti_big_lmm);
2169 OBD_FREE_LARGE(info->eti_big_lmm, info->eti_big_lmmsize);
2170 info->eti_big_lmm = NULL;
2171 info->eti_big_lmmsize = 0;
2173 cl_env_put(env, &refcheck);
2176 #endif /* HAVE_SERVER_SUPPORT */
2178 static int echo_create_object(const struct lu_env *env, struct echo_device *ed,
2181 struct echo_object *eco;
2182 struct echo_client_obd *ec = ed->ed_ec;
2187 if (!(oa->o_valid & OBD_MD_FLID) ||
2188 !(oa->o_valid & OBD_MD_FLGROUP) ||
2189 !fid_seq_is_echo(ostid_seq(&oa->o_oi))) {
2190 CERROR("invalid oid "DOSTID"\n", POSTID(&oa->o_oi));
2194 if (ostid_id(&oa->o_oi) == 0)
2195 ostid_set_id(&oa->o_oi, ++last_object_id);
2197 rc = obd_create(env, ec->ec_exp, oa);
2199 CERROR("Cannot create objects: rc = %d\n", rc);
2205 oa->o_valid |= OBD_MD_FLID;
2207 eco = cl_echo_object_find(ed, &oa->o_oi);
2209 GOTO(failed, rc = PTR_ERR(eco));
2210 cl_echo_object_put(eco);
2212 CDEBUG(D_INFO, "oa oid "DOSTID"\n", POSTID(&oa->o_oi));
2216 if (created && rc != 0)
2217 obd_destroy(env, ec->ec_exp, oa);
2220 CERROR("create object failed with: rc = %d\n", rc);
2225 static int echo_get_object(struct echo_object **ecop, struct echo_device *ed,
2228 struct echo_object *eco;
2232 if (!(oa->o_valid & OBD_MD_FLID) ||
2233 !(oa->o_valid & OBD_MD_FLGROUP) ||
2234 ostid_id(&oa->o_oi) == 0) {
2235 CERROR("invalid oid "DOSTID"\n", POSTID(&oa->o_oi));
2240 eco = cl_echo_object_find(ed, &oa->o_oi);
2249 static void echo_put_object(struct echo_object *eco)
2253 rc = cl_echo_object_put(eco);
2255 CERROR("%s: echo client drop an object failed: rc = %d\n",
2256 eco->eo_dev->ed_ec->ec_exp->exp_obd->obd_name, rc);
2259 static void echo_client_page_debug_setup(struct page *page, int rw, u64 id,
2260 u64 offset, u64 count)
2267 /* no partial pages on the client */
2268 LASSERT(count == PAGE_CACHE_SIZE);
2272 for (delta = 0; delta < PAGE_CACHE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2273 if (rw == OBD_BRW_WRITE) {
2274 stripe_off = offset + delta;
2277 stripe_off = 0xdeadbeef00c0ffeeULL;
2278 stripe_id = 0xdeadbeef00c0ffeeULL;
2280 block_debug_setup(addr + delta, OBD_ECHO_BLOCK_SIZE,
2281 stripe_off, stripe_id);
2288 echo_client_page_debug_check(struct page *page, u64 id, u64 offset, u64 count)
2297 /* no partial pages on the client */
2298 LASSERT(count == PAGE_CACHE_SIZE);
2302 for (rc = delta = 0; delta < PAGE_CACHE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2303 stripe_off = offset + delta;
2306 rc2 = block_debug_check("test_brw",
2307 addr + delta, OBD_ECHO_BLOCK_SIZE,
2308 stripe_off, stripe_id);
2310 CERROR ("Error in echo object "LPX64"\n", id);
2319 static int echo_client_kbrw(struct echo_device *ed, int rw, struct obdo *oa,
2320 struct echo_object *eco, u64 offset,
2321 u64 count, int async)
2324 struct brw_page *pga;
2325 struct brw_page *pgp;
2326 struct page **pages;
2335 verify = (ostid_id(&oa->o_oi) != ECHO_PERSISTENT_OBJID &&
2336 (oa->o_valid & OBD_MD_FLFLAGS) != 0 &&
2337 (oa->o_flags & OBD_FL_DEBUG_CHECK) != 0);
2339 gfp_mask = ((ostid_id(&oa->o_oi) & 2) == 0) ? GFP_IOFS : GFP_HIGHUSER;
2341 LASSERT(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ);
2343 if ((count & (~PAGE_MASK)) != 0)
2346 /* XXX think again with misaligned I/O */
2347 npages = count >> PAGE_CACHE_SHIFT;
2349 if (rw == OBD_BRW_WRITE)
2350 brw_flags = OBD_BRW_ASYNC;
2352 OBD_ALLOC(pga, npages * sizeof(*pga));
2356 OBD_ALLOC(pages, npages * sizeof(*pages));
2357 if (pages == NULL) {
2358 OBD_FREE(pga, npages * sizeof(*pga));
2362 for (i = 0, pgp = pga, off = offset;
2364 i++, pgp++, off += PAGE_CACHE_SIZE) {
2366 LASSERT(pgp->pg == NULL); /* for cleanup */
2369 pgp->pg = alloc_page(gfp_mask);
2370 if (pgp->pg == NULL)
2374 pgp->count = PAGE_CACHE_SIZE;
2376 pgp->flag = brw_flags;
2379 echo_client_page_debug_setup(pgp->pg, rw,
2380 ostid_id(&oa->o_oi), off,
2384 /* brw mode can only be used at client */
2385 LASSERT(ed->ed_next != NULL);
2386 rc = cl_echo_object_brw(eco, rw, offset, pages, npages, async);
2389 if (rc != 0 || rw != OBD_BRW_READ)
2392 for (i = 0, pgp = pga; i < npages; i++, pgp++) {
2393 if (pgp->pg == NULL)
2398 vrc = echo_client_page_debug_check(pgp->pg,
2399 ostid_id(&oa->o_oi),
2400 pgp->off, pgp->count);
2401 if (vrc != 0 && rc == 0)
2404 __free_page(pgp->pg);
2406 OBD_FREE(pga, npages * sizeof(*pga));
2407 OBD_FREE(pages, npages * sizeof(*pages));
2411 static int echo_client_prep_commit(const struct lu_env *env,
2412 struct obd_export *exp, int rw,
2413 struct obdo *oa, struct echo_object *eco,
2414 u64 offset, u64 count,
2415 u64 batch, int async)
2417 struct obd_ioobj ioo;
2418 struct niobuf_local *lnb;
2419 struct niobuf_remote rnb;
2421 u64 npages, tot_pages;
2422 int i, ret = 0, brw_flags = 0;
2426 if (count <= 0 || (count & ~PAGE_CACHE_MASK) != 0)
2429 npages = batch >> PAGE_CACHE_SHIFT;
2430 tot_pages = count >> PAGE_CACHE_SHIFT;
2432 OBD_ALLOC(lnb, npages * sizeof(struct niobuf_local));
2434 GOTO(out, ret = -ENOMEM);
2436 if (rw == OBD_BRW_WRITE && async)
2437 brw_flags |= OBD_BRW_ASYNC;
2439 obdo_to_ioobj(oa, &ioo);
2443 for (; tot_pages > 0; tot_pages -= npages) {
2446 if (tot_pages < npages)
2449 rnb.rnb_offset = off;
2450 rnb.rnb_len = npages * PAGE_CACHE_SIZE;
2451 rnb.rnb_flags = brw_flags;
2453 off += npages * PAGE_CACHE_SIZE;
2456 ret = obd_preprw(env, rw, exp, oa, 1, &ioo, &rnb, &lpages, lnb);
2460 for (i = 0; i < lpages; i++) {
2461 struct page *page = lnb[i].lnb_page;
2463 /* read past eof? */
2464 if (page == NULL && lnb[i].lnb_rc == 0)
2468 lnb[i].lnb_flags |= OBD_BRW_ASYNC;
2470 if (ostid_id(&oa->o_oi) == ECHO_PERSISTENT_OBJID ||
2471 (oa->o_valid & OBD_MD_FLFLAGS) == 0 ||
2472 (oa->o_flags & OBD_FL_DEBUG_CHECK) == 0)
2475 if (rw == OBD_BRW_WRITE)
2476 echo_client_page_debug_setup(page, rw,
2477 ostid_id(&oa->o_oi),
2478 lnb[i].lnb_file_offset,
2481 echo_client_page_debug_check(page,
2482 ostid_id(&oa->o_oi),
2483 lnb[i].lnb_file_offset,
2487 ret = obd_commitrw(env, rw, exp, oa, 1, &ioo, &rnb, npages, lnb,
2492 /* Reuse env context. */
2493 lu_context_exit((struct lu_context *)&env->le_ctx);
2494 lu_context_enter((struct lu_context *)&env->le_ctx);
2499 OBD_FREE(lnb, npages * sizeof(struct niobuf_local));
2503 static int echo_client_brw_ioctl(const struct lu_env *env, int rw,
2504 struct obd_export *exp,
2505 struct obd_ioctl_data *data)
2507 struct obd_device *obd = class_exp2obd(exp);
2508 struct echo_device *ed = obd2echo_dev(obd);
2509 struct echo_client_obd *ec = ed->ed_ec;
2510 struct obdo *oa = &data->ioc_obdo1;
2511 struct echo_object *eco;
2517 LASSERT(oa->o_valid & OBD_MD_FLGROUP);
2519 rc = echo_get_object(&eco, ed, oa);
2523 oa->o_valid &= ~OBD_MD_FLHANDLE;
2525 /* OFD/obdfilter works only via prep/commit */
2526 test_mode = (long)data->ioc_pbuf1;
2527 if (ed->ed_next == NULL && test_mode != 3) {
2529 data->ioc_plen1 = data->ioc_count;
2535 /* Truncate batch size to maximum */
2536 if (data->ioc_plen1 > PTLRPC_MAX_BRW_SIZE)
2537 data->ioc_plen1 = PTLRPC_MAX_BRW_SIZE;
2539 switch (test_mode) {
2543 rc = echo_client_kbrw(ed, rw, oa, eco, data->ioc_offset,
2544 data->ioc_count, async);
2547 rc = echo_client_prep_commit(env, ec->ec_exp, rw, oa, eco,
2548 data->ioc_offset, data->ioc_count,
2549 data->ioc_plen1, async);
2555 echo_put_object(eco);
2561 echo_client_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2562 void *karg, void *uarg)
2564 #ifdef HAVE_SERVER_SUPPORT
2565 struct tgt_session_info *tsi;
2567 struct obd_device *obd = exp->exp_obd;
2568 struct echo_device *ed = obd2echo_dev(obd);
2569 struct echo_client_obd *ec = ed->ed_ec;
2570 struct echo_object *eco;
2571 struct obd_ioctl_data *data = karg;
2575 int rw = OBD_BRW_READ;
2577 #ifdef HAVE_SERVER_SUPPORT
2578 struct lu_context echo_session;
2582 oa = &data->ioc_obdo1;
2583 if (!(oa->o_valid & OBD_MD_FLGROUP)) {
2584 oa->o_valid |= OBD_MD_FLGROUP;
2585 ostid_set_seq_echo(&oa->o_oi);
2588 /* This FID is unpacked just for validation at this point */
2589 rc = ostid_to_fid(&fid, &oa->o_oi, 0);
2597 rc = lu_env_init(env, LCT_DT_THREAD);
2599 GOTO(out_alloc, rc = -ENOMEM);
2601 #ifdef HAVE_SERVER_SUPPORT
2602 env->le_ses = &echo_session;
2603 rc = lu_context_init(env->le_ses, LCT_SERVER_SESSION | LCT_NOREF);
2604 if (unlikely(rc < 0))
2606 lu_context_enter(env->le_ses);
2608 tsi = tgt_ses_info(env);
2609 tsi->tsi_exp = ec->ec_exp;
2610 tsi->tsi_jobid = NULL;
2613 case OBD_IOC_CREATE: /* may create echo object */
2614 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2615 GOTO (out, rc = -EPERM);
2617 rc = echo_create_object(env, ed, oa);
2620 #ifdef HAVE_SERVER_SUPPORT
2621 case OBD_IOC_ECHO_MD: {
2628 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2629 GOTO(out, rc = -EPERM);
2631 count = data->ioc_count;
2632 cmd = data->ioc_command;
2634 id = data->ioc_obdo2.o_oi.oi.oi_id;
2635 dirlen = data->ioc_plen1;
2636 OBD_ALLOC(dir, dirlen + 1);
2638 GOTO(out, rc = -ENOMEM);
2640 if (copy_from_user(dir, data->ioc_pbuf1, dirlen)) {
2641 OBD_FREE(dir, data->ioc_plen1 + 1);
2642 GOTO(out, rc = -EFAULT);
2645 rc = echo_md_handler(ed, cmd, dir, dirlen, id, count, data);
2646 OBD_FREE(dir, dirlen + 1);
2649 case OBD_IOC_ECHO_ALLOC_SEQ: {
2650 struct lu_env *cl_env;
2655 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2656 GOTO(out, rc = -EPERM);
2658 cl_env = cl_env_get(&refcheck);
2660 GOTO(out, rc = PTR_ERR(cl_env));
2662 rc = lu_env_refill_by_tags(cl_env, ECHO_MD_CTX_TAG,
2665 cl_env_put(cl_env, &refcheck);
2669 rc = seq_client_get_seq(cl_env, ed->ed_cl_seq, &seq);
2670 cl_env_put(cl_env, &refcheck);
2672 CERROR("%s: Can not alloc seq: rc = %d\n",
2677 if (copy_to_user(data->ioc_pbuf1, &seq, data->ioc_plen1))
2680 max_count = LUSTRE_METADATA_SEQ_MAX_WIDTH;
2681 if (copy_to_user(data->ioc_pbuf2, &max_count,
2686 #endif /* HAVE_SERVER_SUPPORT */
2687 case OBD_IOC_DESTROY:
2688 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2689 GOTO (out, rc = -EPERM);
2691 rc = echo_get_object(&eco, ed, oa);
2693 rc = obd_destroy(env, ec->ec_exp, oa);
2695 eco->eo_deleted = 1;
2696 echo_put_object(eco);
2700 case OBD_IOC_GETATTR:
2701 rc = echo_get_object(&eco, ed, oa);
2703 rc = obd_getattr(env, ec->ec_exp, oa);
2704 echo_put_object(eco);
2708 case OBD_IOC_SETATTR:
2709 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2710 GOTO (out, rc = -EPERM);
2712 rc = echo_get_object(&eco, ed, oa);
2714 rc = obd_setattr(env, ec->ec_exp, oa);
2715 echo_put_object(eco);
2719 case OBD_IOC_BRW_WRITE:
2720 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2721 GOTO (out, rc = -EPERM);
2725 case OBD_IOC_BRW_READ:
2726 rc = echo_client_brw_ioctl(env, rw, exp, data);
2730 CERROR ("echo_ioctl(): unrecognised ioctl %#x\n", cmd);
2731 GOTO (out, rc = -ENOTTY);
2736 #ifdef HAVE_SERVER_SUPPORT
2737 lu_context_exit(env->le_ses);
2738 lu_context_fini(env->le_ses);
2748 static int echo_client_setup(const struct lu_env *env,
2749 struct obd_device *obddev, struct lustre_cfg *lcfg)
2751 struct echo_client_obd *ec = &obddev->u.echo_client;
2752 struct obd_device *tgt;
2753 struct obd_uuid echo_uuid = { "ECHO_UUID" };
2754 struct obd_connect_data *ocd = NULL;
2758 if (lcfg->lcfg_bufcount < 2 || LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
2759 CERROR("requires a TARGET OBD name\n");
2763 tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
2764 if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
2765 CERROR("device not attached or not set up (%s)\n",
2766 lustre_cfg_string(lcfg, 1));
2770 spin_lock_init(&ec->ec_lock);
2771 INIT_LIST_HEAD(&ec->ec_objects);
2772 INIT_LIST_HEAD(&ec->ec_locks);
2775 if (!strcmp(tgt->obd_type->typ_name, LUSTRE_MDT_NAME)) {
2776 #ifdef HAVE_SERVER_SUPPORT
2777 lu_context_tags_update(ECHO_MD_CTX_TAG);
2778 lu_session_tags_update(ECHO_MD_SES_TAG);
2780 CERROR("Local operations are NOT supported on client side. "
2781 "Only remote operations are supported. Metadata client "
2782 "must be run on server side.\n");
2787 OBD_ALLOC(ocd, sizeof(*ocd));
2789 CERROR("Can't alloc ocd connecting to %s\n",
2790 lustre_cfg_string(lcfg, 1));
2794 ocd->ocd_connect_flags = OBD_CONNECT_VERSION | OBD_CONNECT_REQPORTAL |
2795 OBD_CONNECT_BRW_SIZE |
2796 OBD_CONNECT_GRANT | OBD_CONNECT_FULL20 |
2797 OBD_CONNECT_64BITHASH | OBD_CONNECT_LVB_TYPE |
2799 ocd->ocd_brw_size = DT_MAX_BRW_SIZE;
2800 ocd->ocd_version = LUSTRE_VERSION_CODE;
2801 ocd->ocd_group = FID_SEQ_ECHO;
2803 rc = obd_connect(env, &ec->ec_exp, tgt, &echo_uuid, ocd, NULL);
2805 /* Turn off pinger because it connects to tgt obd directly. */
2806 spin_lock(&tgt->obd_dev_lock);
2807 list_del_init(&ec->ec_exp->exp_obd_chain_timed);
2808 spin_unlock(&tgt->obd_dev_lock);
2811 OBD_FREE(ocd, sizeof(*ocd));
2814 CERROR("fail to connect to device %s\n",
2815 lustre_cfg_string(lcfg, 1));
2822 static int echo_client_cleanup(struct obd_device *obddev)
2824 struct echo_device *ed = obd2echo_dev(obddev);
2825 struct echo_client_obd *ec = &obddev->u.echo_client;
2829 /*Do nothing for Metadata echo client*/
2833 if (ed->ed_next_ismd) {
2834 #ifdef HAVE_SERVER_SUPPORT
2835 lu_context_tags_clear(ECHO_MD_CTX_TAG);
2836 lu_session_tags_clear(ECHO_MD_SES_TAG);
2838 CERROR("This is client-side only module, does not support "
2839 "metadata echo client.\n");
2844 if (!list_empty(&obddev->obd_exports)) {
2845 CERROR("still has clients!\n");
2849 LASSERT(atomic_read(&ec->ec_exp->exp_refcount) > 0);
2850 rc = obd_disconnect(ec->ec_exp);
2852 CERROR("fail to disconnect device: %d\n", rc);
2857 static int echo_client_connect(const struct lu_env *env,
2858 struct obd_export **exp,
2859 struct obd_device *src, struct obd_uuid *cluuid,
2860 struct obd_connect_data *data, void *localdata)
2863 struct lustre_handle conn = { 0 };
2866 rc = class_connect(&conn, src, cluuid);
2868 *exp = class_conn2export(&conn);
2874 static int echo_client_disconnect(struct obd_export *exp)
2880 GOTO(out, rc = -EINVAL);
2882 rc = class_disconnect(exp);
2888 static struct obd_ops echo_client_obd_ops = {
2889 .o_owner = THIS_MODULE,
2890 .o_iocontrol = echo_client_iocontrol,
2891 .o_connect = echo_client_connect,
2892 .o_disconnect = echo_client_disconnect
2895 static int echo_client_init(void)
2899 rc = lu_kmem_init(echo_caches);
2901 rc = class_register_type(&echo_client_obd_ops, NULL, true, NULL,
2902 LUSTRE_ECHO_CLIENT_NAME,
2905 lu_kmem_fini(echo_caches);
2910 static void echo_client_exit(void)
2912 class_unregister_type(LUSTRE_ECHO_CLIENT_NAME);
2913 lu_kmem_fini(echo_caches);
2916 static int __init obdecho_init(void)
2921 LCONSOLE_INFO("Echo OBD driver; http://www.lustre.org/\n");
2923 LASSERT(PAGE_CACHE_SIZE % OBD_ECHO_BLOCK_SIZE == 0);
2925 # ifdef HAVE_SERVER_SUPPORT
2926 rc = echo_persistent_pages_init();
2930 rc = class_register_type(&echo_obd_ops, NULL, true, NULL,
2931 LUSTRE_ECHO_NAME, NULL);
2936 rc = echo_client_init();
2938 # ifdef HAVE_SERVER_SUPPORT
2942 class_unregister_type(LUSTRE_ECHO_NAME);
2944 echo_persistent_pages_fini();
2950 static void /*__exit*/ obdecho_exit(void)
2954 # ifdef HAVE_SERVER_SUPPORT
2955 class_unregister_type(LUSTRE_ECHO_NAME);
2956 echo_persistent_pages_fini();
2960 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
2961 MODULE_DESCRIPTION("Lustre Testing Echo OBD driver");
2962 MODULE_VERSION(LUSTRE_VERSION_STRING);
2963 MODULE_LICENSE("GPL");
2965 module_init(obdecho_init);
2966 module_exit(obdecho_exit);
2968 /** @} echo_client */