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.gnu.org/licenses/gpl-2.0.html
23 * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24 * Use is subject to license terms.
26 * Copyright (c) 2011, 2017, Intel Corporation.
29 * This file is part of Lustre, http://www.lustre.org/
30 * Lustre is a trademark of Sun Microsystems, Inc.
33 #define DEBUG_SUBSYSTEM S_ECHO
35 #include <linux/user_namespace.h>
36 #include <linux/uidgid.h>
38 #include <libcfs/libcfs.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>
53 #define ETI_NAME_LEN 20
55 #endif /* HAVE_SERVER_SUPPORT */
57 #include "echo_internal.h"
59 /** \defgroup echo_client Echo Client
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)
69 struct cl_device ed_cl;
70 struct echo_client_obd *ed_ec;
72 struct cl_site ed_site_myself;
73 struct lu_site *ed_site;
74 struct lu_device *ed_next;
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 */
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;
93 struct echo_object_conf {
94 struct cl_object_conf eoc_cl;
95 struct lov_oinfo **eoc_oinfo;
99 struct cl_page_slice ep_cl;
100 unsigned long ep_lock;
104 struct cl_lock_slice el_cl;
105 struct list_head el_chain;
106 struct echo_object *el_object;
108 atomic_t el_refcount;
111 #ifdef HAVE_SERVER_SUPPORT
112 static const char echo_md_root_dir_name[] = "ROOT_ECHO";
115 * In order to use the values of members in struct mdd_device,
116 * we define an alias structure here.
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;
126 #endif /* HAVE_SERVER_SUPPORT */
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);
133 /** \defgroup echo_helpers Helper functions
136 static inline struct echo_device *cl2echo_dev(const struct cl_device *dev)
138 return container_of0(dev, struct echo_device, ed_cl);
141 static inline struct cl_device *echo_dev2cl(struct echo_device *d)
146 static inline struct echo_device *obd2echo_dev(const struct obd_device *obd)
148 return cl2echo_dev(lu2cl_dev(obd->obd_lu_dev));
151 static inline struct cl_object *echo_obj2cl(struct echo_object *eco)
156 static inline struct echo_object *cl2echo_obj(const struct cl_object *o)
158 return container_of(o, struct echo_object, eo_cl);
161 static inline struct echo_page *cl2echo_page(const struct cl_page_slice *s)
163 return container_of(s, struct echo_page, ep_cl);
166 static inline struct echo_lock *cl2echo_lock(const struct cl_lock_slice *s)
168 return container_of(s, struct echo_lock, el_cl);
171 static inline struct cl_lock *echo_lock2cl(const struct echo_lock *ecl)
173 return ecl->el_cl.cls_lock;
176 static struct lu_context_key echo_thread_key;
178 static inline struct echo_thread_info *echo_env_info(const struct lu_env *env)
180 struct echo_thread_info *info;
182 info = lu_context_key_get(&env->le_ctx, &echo_thread_key);
183 LASSERT(info != NULL);
188 struct echo_object_conf *cl2echo_conf(const struct cl_object_conf *c)
190 return container_of(c, struct echo_object_conf, eoc_cl);
193 #ifdef HAVE_SERVER_SUPPORT
194 static inline struct echo_md_device *lu2emd_dev(struct lu_device *d)
196 return container_of0(d, struct echo_md_device, emd_md_dev.md_lu_dev);
199 static inline struct lu_device *emd2lu_dev(struct echo_md_device *d)
201 return &d->emd_md_dev.md_lu_dev;
204 static inline struct seq_server_site *echo_md_seq_site(struct echo_md_device *d)
206 return emd2lu_dev(d)->ld_site->ld_seq_site;
209 static inline struct obd_device *emd2obd_dev(struct echo_md_device *d)
211 return d->emd_md_dev.md_lu_dev.ld_obd;
213 #endif /* HAVE_SERVER_SUPPORT */
215 /** @} echo_helpers */
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);
221 struct echo_thread_info {
222 struct echo_object_conf eti_conf;
223 struct lustre_md eti_md;
224 struct cl_2queue eti_queue;
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 */
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];
245 /* No session used right now */
246 struct echo_session_info {
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; */
256 static struct lu_kmem_descr echo_caches[] = {
258 .ckd_cache = &echo_lock_kmem,
259 .ckd_name = "echo_lock_kmem",
260 .ckd_size = sizeof(struct echo_lock)
263 .ckd_cache = &echo_object_kmem,
264 .ckd_name = "echo_object_kmem",
265 .ckd_size = sizeof(struct echo_object)
268 .ckd_cache = &echo_thread_kmem,
269 .ckd_name = "echo_thread_kmem",
270 .ckd_size = sizeof(struct echo_thread_info)
273 .ckd_cache = &echo_session_kmem,
274 .ckd_name = "echo_session_kmem",
275 .ckd_size = sizeof(struct echo_session_info)
282 /** \defgroup echo_page Page operations
284 * Echo page operations.
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)
292 struct echo_page *ep = cl2echo_page(slice);
295 if (test_and_set_bit(0, &ep->ep_lock))
298 while (test_and_set_bit(0, &ep->ep_lock))
299 wait_on_bit(&ep->ep_lock, 0, TASK_UNINTERRUPTIBLE);
304 static void echo_page_disown(const struct lu_env *env,
305 const struct cl_page_slice *slice,
308 struct echo_page *ep = cl2echo_page(slice);
310 LASSERT(test_bit(0, &ep->ep_lock));
311 clear_and_wake_up_bit(0, &ep->ep_lock);
314 static void echo_page_discard(const struct lu_env *env,
315 const struct cl_page_slice *slice,
316 struct cl_io *unused)
318 cl_page_delete(env, slice->cpl_page);
321 static int echo_page_is_vmlocked(const struct lu_env *env,
322 const struct cl_page_slice *slice)
324 if (test_bit(0, &cl2echo_page(slice)->ep_lock))
329 static void echo_page_completion(const struct lu_env *env,
330 const struct cl_page_slice *slice,
333 LASSERT(slice->cpl_page->cp_sync_io != NULL);
336 static void echo_page_fini(const struct lu_env *env,
337 struct cl_page_slice *slice,
338 struct pagevec *pvec)
340 struct echo_object *eco = cl2echo_obj(slice->cpl_obj);
343 atomic_dec(&eco->eo_npages);
344 put_page(slice->cpl_page->cp_vmpage);
348 static int echo_page_prep(const struct lu_env *env,
349 const struct cl_page_slice *slice,
350 struct cl_io *unused)
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)
359 struct echo_page *ep = cl2echo_page(slice);
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);
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,
376 .cpo_prep = echo_page_prep,
377 .cpo_completion = echo_page_completion,
380 .cpo_prep = echo_page_prep,
381 .cpo_completion = echo_page_completion,
388 /** \defgroup echo_lock Locking
390 * echo lock operations
394 static void echo_lock_fini(const struct lu_env *env,
395 struct cl_lock_slice *slice)
397 struct echo_lock *ecl = cl2echo_lock(slice);
399 LASSERT(list_empty(&ecl->el_chain));
400 OBD_SLAB_FREE_PTR(ecl, echo_lock_kmem);
403 static struct cl_lock_operations echo_lock_ops = {
404 .clo_fini = echo_lock_fini,
409 /** \defgroup echo_cl_ops cl_object operations
411 * operations for cl_object
415 static int echo_page_init(const struct lu_env *env, struct cl_object *obj,
416 struct cl_page *page, pgoff_t index)
418 struct echo_page *ep = cl_object_page_slice(obj, page);
419 struct echo_object *eco = cl2echo_obj(obj);
422 get_page(page->cp_vmpage);
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.
430 cl_page_slice_add(page, &ep->ep_cl, obj, index, &echo_page_ops);
431 atomic_inc(&eco->eo_npages);
435 static int echo_io_init(const struct lu_env *env, struct cl_object *obj,
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)
445 struct echo_lock *el;
448 OBD_SLAB_ALLOC_PTR_GFP(el, echo_lock_kmem, GFP_NOFS);
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);
455 RETURN(el ? 0 : -ENOMEM);
458 static int echo_conf_set(const struct lu_env *env, struct cl_object *obj,
459 const struct cl_object_conf *conf)
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
470 /** @} echo_cl_ops */
472 /** \defgroup echo_lu_ops lu_object operations
474 * operations for echo lu object.
478 static int echo_object_init(const struct lu_env *env, struct lu_object *obj,
479 const struct lu_object_conf *conf)
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));
487 struct lu_object *below;
488 struct lu_device *under;
491 below = under->ld_ops->ldo_object_alloc(env, obj->lo_header,
495 lu_object_add(obj, below);
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);
502 LASSERT(econf->eoc_oinfo != NULL);
505 * Transfer the oinfo pointer to eco that it won't be
508 eco->eo_oinfo = *econf->eoc_oinfo;
509 *econf->eoc_oinfo = NULL;
511 eco->eo_oinfo = NULL;
515 atomic_set(&eco->eo_npages, 0);
516 cl_object_page_init(lu2cl(obj), sizeof(struct echo_page));
518 spin_lock(&ec->ec_lock);
519 list_add_tail(&eco->eo_obj_chain, &ec->ec_objects);
520 spin_unlock(&ec->ec_lock);
525 static void echo_object_delete(const struct lu_env *env, struct lu_object *obj)
527 struct echo_object *eco = cl2echo_obj(lu2cl(obj));
528 struct echo_client_obd *ec;
532 /* object delete called unconditolally - layer init or not */
533 if (eco->eo_dev == NULL)
536 ec = eco->eo_dev->ed_ec;
538 LASSERT(atomic_read(&eco->eo_npages) == 0);
540 spin_lock(&ec->ec_lock);
541 list_del_init(&eco->eo_obj_chain);
542 spin_unlock(&ec->ec_lock);
545 OBD_FREE_PTR(eco->eo_oinfo);
548 static void echo_object_free(const struct lu_env *env, struct lu_object *obj)
550 struct echo_object *eco = cl2echo_obj(lu2cl(obj));
555 lu_object_header_fini(obj->lo_header);
557 OBD_SLAB_FREE_PTR(eco, echo_object_kmem);
561 static int echo_object_print(const struct lu_env *env, void *cookie,
562 lu_printer_t p, const struct lu_object *o)
564 struct echo_object *obj = cl2echo_obj(lu2cl(o));
566 return (*p)(env, cookie, "echoclient-object@%p", obj);
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
577 /** @} echo_lu_ops */
579 /** \defgroup echo_lu_dev_ops lu_device operations
581 * Operations for echo lu device.
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)
589 struct echo_object *eco;
590 struct lu_object *obj = NULL;
593 /* we're the top dev. */
594 LASSERT(hdr == NULL);
595 OBD_SLAB_ALLOC_PTR_GFP(eco, echo_object_kmem, GFP_NOFS);
597 struct cl_object_header *hdr = &eco->eo_hdr;
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));
603 lu_object_init(obj, &hdr->coh_lu, dev);
604 lu_object_add_top(&hdr->coh_lu, obj);
606 eco->eo_cl.co_ops = &echo_cl_obj_ops;
607 obj->lo_ops = &echo_lu_obj_ops;
612 static struct lu_device_operations echo_device_lu_ops = {
613 .ldo_object_alloc = echo_object_alloc,
616 /** @} echo_lu_dev_ops */
618 /** \defgroup echo_init Setup and teardown
620 * Init and fini functions for echo client.
624 static int echo_site_init(const struct lu_env *env, struct echo_device *ed)
626 struct cl_site *site = &ed->ed_site_myself;
629 /* initialize site */
630 rc = cl_site_init(site, &ed->ed_cl);
632 CERROR("Cannot initialize site for echo client(%d)\n", rc);
636 rc = lu_site_init_finish(&site->cs_lu);
642 ed->ed_site = &site->cs_lu;
646 static void echo_site_fini(const struct lu_env *env, struct echo_device *ed)
649 if (!ed->ed_next_ismd)
650 lu_site_fini(ed->ed_site);
655 static void *echo_thread_key_init(const struct lu_context *ctx,
656 struct lu_context_key *key)
658 struct echo_thread_info *info;
660 OBD_SLAB_ALLOC_PTR_GFP(info, echo_thread_kmem, GFP_NOFS);
662 info = ERR_PTR(-ENOMEM);
666 static void echo_thread_key_fini(const struct lu_context *ctx,
667 struct lu_context_key *key, void *data)
669 struct echo_thread_info *info = data;
671 OBD_SLAB_FREE_PTR(info, echo_thread_kmem);
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,
680 static void *echo_session_key_init(const struct lu_context *ctx,
681 struct lu_context_key *key)
683 struct echo_session_info *session;
685 OBD_SLAB_ALLOC_PTR_GFP(session, echo_session_kmem, GFP_NOFS);
687 session = ERR_PTR(-ENOMEM);
691 static void echo_session_key_fini(const struct lu_context *ctx,
692 struct lu_context_key *key, void *data)
694 struct echo_session_info *session = data;
696 OBD_SLAB_FREE_PTR(session, echo_session_kmem);
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,
705 LU_TYPE_INIT_FINI(echo, &echo_thread_key, &echo_session_key);
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)
716 OBD_ALLOC_PTR(ed->ed_cl_seq);
720 OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
722 GOTO(out_free_seq, rc = -ENOMEM);
724 snprintf(prefix, MAX_OBD_NAME + 5, "srv-%s", obd_name);
726 /* Init client side sequence-manager */
727 rc = seq_client_init(ed->ed_cl_seq, NULL,
729 prefix, ss->ss_server_seq);
730 ed->ed_cl_seq->lcs_width = ECHO_SEQ_WIDTH;
731 OBD_FREE(prefix, MAX_OBD_NAME + 5);
733 GOTO(out_free_seq, rc);
738 OBD_FREE_PTR(ed->ed_cl_seq);
739 ed->ed_cl_seq = NULL;
743 static int echo_fid_fini(struct obd_device *obddev)
745 struct echo_device *ed = obd2echo_dev(obddev);
749 seq_client_fini(ed->ed_cl_seq);
750 OBD_FREE_PTR(ed->ed_cl_seq);
751 ed->ed_cl_seq = NULL;
757 static void echo_ed_los_fini(const struct lu_env *env, struct echo_device *ed)
760 if (ed != NULL && ed->ed_next_ismd && ed->ed_los != NULL) {
761 local_oid_storage_fini(env, ed->ed_los);
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)
772 struct dt_object *parent = NULL;
773 struct dt_object *dto = NULL;
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));
782 /* create local file with @fid */
783 dto = local_file_find_or_create_with_fid(env, emd->emd_bottom, fid,
786 GOTO(out_put, rc = PTR_ERR(dto));
788 *fid = *lu_object_fid(&dto->do_lu);
790 * since stack is not fully set up the local_storage uses own stack
791 * and we should drop its object from cache
793 dt_object_put_nocache(env, dto);
797 dt_object_put(env, parent);
802 echo_md_root_get(const struct lu_env *env, struct echo_md_device *emd,
803 struct echo_device *ed)
809 /* Setup local dirs */
810 fid.f_seq = FID_SEQ_LOCAL_NAME;
813 rc = local_oid_storage_init(env, emd->emd_bottom, &fid, &ed->ed_los);
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,
825 CERROR("%s: create md echo root fid failed: rc = %d\n",
826 emd2obd_dev(emd)->obd_name, rc);
830 ed->ed_root_fid = fid;
834 echo_ed_los_fini(env, ed);
838 #endif /* HAVE_SERVER_SUPPORT */
840 static struct lu_device *echo_device_alloc(const struct lu_env *env,
841 struct lu_device_type *t,
842 struct lustre_cfg *cfg)
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;
856 GOTO(out, rc = -ENOMEM);
860 rc = cl_device_init(cd, t);
864 cd->cd_lu_dev.ld_ops = &echo_device_lu_ops;
867 obd = class_name2obd(lustre_cfg_string(cfg, 0));
868 LASSERT(obd != NULL);
869 LASSERT(env != NULL);
871 tgt = class_name2obd(lustre_cfg_string(cfg, 1));
873 CERROR("Can not find tgt device %s\n",
874 lustre_cfg_string(cfg, 1));
875 GOTO(out, rc = -ENODEV);
878 next = tgt->obd_lu_dev;
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);
889 GOTO(out, rc = -EINVAL);
894 rc = echo_client_setup(env, obd, cfg);
898 ed->ed_ec = &obd->u.echo_client;
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;
911 CERROR("%s is not lu device type!\n",
912 lustre_cfg_string(cfg, 1));
913 GOTO(out, rc = -EINVAL);
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);
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) {
933 spin_unlock(&ls->ls_ld_lock);
936 CERROR("%s is not lu device type!\n",
937 lustre_cfg_string(cfg, 1));
938 GOTO(out, rc = -EINVAL);
942 /* For MD echo client, it will use the site in MDS stack */
944 ed->ed_cl.cd_lu_dev.ld_site = ls;
945 rc = echo_fid_init(ed, obd->obd_name, lu_site2seq(ls));
947 CERROR("echo fid init error %d\n", rc);
951 md = lu2md_dev(next);
952 emd = lu2emd_dev(&md->md_lu_dev);
953 rc = echo_md_root_get(env, emd, ed);
955 CERROR("%s: get root error: rc = %d\n",
956 emd2obd_dev(emd)->obd_name, rc);
959 #else /* !HAVE_SERVER_SUPPORT */
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 */
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
969 if (next != NULL && !lu_device_is_cl(next))
972 tgt_type_name = tgt->obd_type->typ_name;
974 LASSERT(next != NULL);
976 GOTO(out, rc = -EBUSY);
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,
985 LASSERT(strcmp(tgt_type_name, LUSTRE_OST_NAME) == 0);
990 RETURN(&cd->cd_lu_dev);
996 rc2 = echo_client_cleanup(obd);
998 CERROR("Cleanup obd device %s error(%d)\n",
1004 echo_site_fini(env, ed);
1007 cl_device_fini(&ed->ed_cl);
1019 static int echo_device_init(const struct lu_env *env, struct lu_device *d,
1020 const char *name, struct lu_device *next)
1026 static struct lu_device *echo_device_fini(const struct lu_env *env,
1027 struct lu_device *d)
1029 struct echo_device *ed = cl2echo_dev(lu2cl_dev(d));
1030 struct lu_device *next = ed->ed_next;
1032 while (next && !ed->ed_next_ismd)
1033 next = next->ld_type->ldt_ops->ldto_device_fini(env, next);
1037 static void echo_lock_release(const struct lu_env *env,
1038 struct echo_lock *ecl,
1041 struct cl_lock *clk = echo_lock2cl(ecl);
1043 cl_lock_release(env, clk);
1046 static struct lu_device *echo_device_free(const struct lu_env *env,
1047 struct lu_device *d)
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;
1054 CDEBUG(D_INFO, "echo device:%p is going to be freed, next = %p\n",
1057 lu_site_purge(env, ed->ed_site, -1);
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.
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);
1071 lu_site_purge(env, ed->ed_site, -1);
1074 "Waiting for the reference of echo object to be dropped\n");
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);
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);
1087 spin_unlock(&ec->ec_lock);
1089 LASSERT(list_empty(&ec->ec_locks));
1091 CDEBUG(D_INFO, "No object exists, exiting...\n");
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);
1098 while (next && !ed->ed_next_ismd)
1099 next = next->ld_type->ldt_ops->ldto_device_free(env, next);
1101 LASSERT(ed->ed_site == d->ld_site);
1102 echo_site_fini(env, ed);
1103 cl_device_fini(&ed->ed_cl);
1106 cl_env_cache_purge(~0);
1111 static const struct lu_device_type_operations echo_device_type_ops = {
1112 .ldto_init = echo_type_init,
1113 .ldto_fini = echo_type_fini,
1115 .ldto_start = echo_type_start,
1116 .ldto_stop = echo_type_stop,
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
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,
1132 /** \defgroup echo_exports Exported operations
1134 * exporting functions to echo client
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)
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;
1154 LASSERTF(ostid_id(oi) != 0, DOSTID"\n", POSTID(oi));
1155 LASSERTF(ostid_seq(oi) == FID_SEQ_ECHO, DOSTID"\n", POSTID(oi));
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));
1161 env = cl_env_get(&refcheck);
1163 RETURN((void *)env);
1165 info = echo_env_info(env);
1166 conf = &info->eti_conf;
1168 OBD_ALLOC_PTR(oinfo);
1170 GOTO(out, eco = ERR_PTR(-ENOMEM));
1172 oinfo->loi_oi = *oi;
1173 conf->eoc_cl.u.coc_oinfo = oinfo;
1177 * If echo_object_init() is successful then ownership of oinfo
1178 * is transferred to the object.
1180 conf->eoc_oinfo = &oinfo;
1182 fid = &info->eti_fid;
1183 rc = ostid_to_fid(fid, oi, 0);
1185 GOTO(out, eco = ERR_PTR(rc));
1188 * In the function below, .hs_keycmp resolves to
1189 * lu_obj_hop_keycmp()
1191 /* coverity[overrun-buffer-val] */
1192 obj = cl_object_find(env, echo_dev2cl(d), fid, &conf->eoc_cl);
1194 GOTO(out, eco = (void *)obj);
1196 eco = cl2echo_obj(obj);
1197 if (eco->eo_deleted) {
1198 cl_object_put(env, obj);
1199 eco = ERR_PTR(-EAGAIN);
1204 OBD_FREE_PTR(oinfo);
1206 cl_env_put(env, &refcheck);
1210 static int cl_echo_object_put(struct echo_object *eco)
1213 struct cl_object *obj = echo_obj2cl(eco);
1217 env = cl_env_get(&refcheck);
1219 RETURN(PTR_ERR(env));
1221 /* an external function to kill an object? */
1222 if (eco->eo_deleted) {
1223 struct lu_object_header *loh = obj->co_lu.lo_header;
1225 LASSERT(&eco->eo_hdr == luh2coh(loh));
1226 set_bit(LU_OBJECT_HEARD_BANSHEE, &loh->loh_flags);
1229 cl_object_put(env, obj);
1230 cl_env_put(env, &refcheck);
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)
1239 struct cl_lock *lck;
1240 struct cl_object *obj;
1241 struct cl_lock_descr *descr;
1242 struct echo_thread_info *info;
1246 info = echo_env_info(env);
1248 lck = &info->eti_lock;
1249 obj = echo_obj2cl(eco);
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;
1260 rc = cl_lock_request(env, io, lck);
1262 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
1263 struct echo_lock *el;
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;
1271 atomic_inc(&el->el_refcount);
1272 *cookie = el->el_cookie;
1273 spin_unlock(&ec->ec_lock);
1278 static int cl_echo_cancel0(struct lu_env *env, struct echo_device *ed,
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;
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);
1294 if (atomic_dec_and_test(&ecl->el_refcount))
1295 list_del_init(&ecl->el_chain);
1301 spin_unlock(&ec->ec_lock);
1306 echo_lock_release(env, ecl, still_used);
1310 static void echo_commit_callback(const struct lu_env *env, struct cl_io *io,
1311 struct pagevec *pvec)
1313 struct echo_thread_info *info;
1314 struct cl_2queue *queue;
1317 info = echo_env_info(env);
1318 LASSERT(io == &info->eti_io);
1320 queue = &info->eti_queue;
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;
1326 cl_page_list_add(&queue->c2_qout, page);
1330 static int cl_echo_object_brw(struct echo_object *eco, int rw, u64 offset,
1331 struct page **pages, int npages, int async)
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;
1339 struct cl_page *clp;
1340 struct lustre_handle lh = { 0 };
1341 int page_size = cl_page_size(obj);
1347 LASSERT((offset & ~PAGE_MASK) == 0);
1348 LASSERT(ed->ed_next != NULL);
1349 env = cl_env_get(&refcheck);
1351 RETURN(PTR_ERR(env));
1353 info = echo_env_info(env);
1355 queue = &info->eti_queue;
1357 cl_2queue_init(queue);
1359 io->ci_ignore_layout = 1;
1360 rc = cl_io_init(env, io, CIT_MISC, obj);
1365 rc = cl_echo_enqueue0(env, eco, offset,
1366 offset + npages * PAGE_SIZE - 1,
1367 rw == READ ? LCK_PR : LCK_PW, &lh.cookie,
1370 GOTO(error_lock, rc);
1372 for (i = 0; i < npages; i++) {
1374 clp = cl_page_find(env, obj, cl_index(obj, offset),
1375 pages[i], CPT_TRANSIENT);
1380 LASSERT(clp->cp_type == CPT_TRANSIENT);
1382 rc = cl_page_own(env, io, clp);
1384 LASSERT(clp->cp_state == CPS_FREEING);
1385 cl_page_put(env, clp);
1389 cl_2queue_add(queue, clp);
1392 * drop the reference count for cl_page_find, so that the page
1393 * will be freed in cl_2queue_fini.
1395 cl_page_put(env, clp);
1396 cl_page_clip(env, clp, 0, page_size);
1398 offset += page_size;
1402 enum cl_req_type typ = rw == READ ? CRT_READ : CRT_WRITE;
1404 async = async && (typ == CRT_WRITE);
1406 rc = cl_io_commit_async(env, io, &queue->c2_qin,
1408 echo_commit_callback);
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);
1415 cl_echo_cancel0(env, ed, lh.cookie);
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);
1423 cl_env_put(env, &refcheck);
1426 /** @} echo_exports */
1428 static u64 last_object_id;
1430 #ifdef HAVE_SERVER_SUPPORT
1431 static inline void echo_md_build_name(struct lu_name *lname, char *name,
1434 snprintf(name, ETI_NAME_LEN, "%llu", id);
1435 lname->ln_name = name;
1436 lname->ln_namelen = strlen(name);
1439 /* similar to mdt_attr_get_complex */
1440 static int echo_big_lmm_get(const struct lu_env *env, struct md_object *o,
1443 struct echo_thread_info *info = echo_env_info(env);
1448 LASSERT(ma->ma_lmm_size > 0);
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);
1454 rc = mo_xattr_get(env, o, &LU_BUF_NULL, XATTR_NAME_LMV);
1459 /* big_lmm may need to be grown */
1460 if (info->eti_big_lmmsize < rc) {
1461 int size = size_roundup_power2(rc);
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;
1472 OBD_ALLOC_LARGE(info->eti_big_lmm, size);
1473 if (!info->eti_big_lmm)
1475 info->eti_big_lmmsize = size;
1477 LASSERT(info->eti_big_lmmsize >= rc);
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);
1484 rc = mo_xattr_get(env, o, &info->eti_buf, XATTR_NAME_LMV);
1488 if (ma->ma_need & MA_LOV)
1489 ma->ma_valid |= MA_LOV;
1491 ma->ma_valid |= MA_LMV;
1493 ma->ma_lmm = info->eti_big_lmm;
1494 ma->ma_lmm_size = rc;
1499 static int echo_attr_get_complex(const struct lu_env *env,
1500 struct md_object *next,
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);
1512 if (ma->ma_need & MA_INODE) {
1513 rc = mo_attr_get(env, next, ma);
1516 ma->ma_valid |= MA_INODE;
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);
1525 ma->ma_lmm_size = rc2;
1526 ma->ma_valid |= MA_LOV;
1527 } else if (rc2 == -ENODATA) {
1529 ma->ma_lmm_size = 0;
1530 } else if (rc2 == -ERANGE) {
1531 rc2 = echo_big_lmm_get(env, next, ma);
1533 GOTO(out, rc = rc2);
1535 GOTO(out, rc = rc2);
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);
1545 ma->ma_lmm_size = rc2;
1546 ma->ma_valid |= MA_LMV;
1547 } else if (rc2 == -ENODATA) {
1549 ma->ma_lmm_size = 0;
1550 } else if (rc2 == -ERANGE) {
1551 rc2 = echo_big_lmm_get(env, next, ma);
1553 GOTO(out, rc = rc2);
1555 GOTO(out, rc = rc2);
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);
1565 ma->ma_acl_size = rc2;
1566 ma->ma_valid |= MA_ACL_DEF;
1567 } else if (rc2 == -ENODATA) {
1569 ma->ma_acl_size = 0;
1571 GOTO(out, rc = rc2);
1576 CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = %#llx ma_lmm=%p\n",
1577 rc, ma->ma_valid, ma->ma_lmm);
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,
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 };
1596 rc = mdo_lookup(env, parent, lname, fid2, spec);
1599 else if (rc != -ENOENT)
1602 ec_child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev,
1604 if (IS_ERR(ec_child)) {
1605 CERROR("Can not find the child "DFID": rc = %ld\n", PFID(fid),
1607 RETURN(PTR_ERR(ec_child));
1610 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1612 CERROR("Can not locate the child "DFID"\n", PFID(fid));
1613 GOTO(out_put, rc = -EINVAL);
1616 CDEBUG(D_RPCTRACE, "Start creating object "DFID" %s %p\n",
1617 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
1620 * Do not perform lookup sanity check. We know that name does not exist.
1622 spec->sp_cr_lookup = 0;
1623 rc = mdo_create(env, parent, lname, lu2md(child), spec, ma);
1625 CERROR("Can not create child "DFID": rc = %d\n", PFID(fid), rc);
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);
1632 lu_object_put(env, ec_child);
1636 static int echo_set_lmm_size(const struct lu_env *env, struct lu_device *ld,
1639 struct echo_thread_info *info = echo_env_info(env);
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);
1645 LASSERT(info->eti_big_lmmsize);
1646 ma->ma_lmm = info->eti_big_lmm;
1647 ma->ma_lmm_size = info->eti_big_lmmsize;
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)
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;
1664 struct lu_name tmp_ln_name;
1665 struct lu_fid stripe_fid;
1666 struct lu_object *stripe_obj;
1669 LASSERT(obj != NULL);
1670 LASSERT(S_ISDIR(obj->lo_header->loh_attr));
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);
1677 CERROR("Can not getattr child "DFID": rc = %d\n",
1678 PFID(lu_object_fid(obj)), rc);
1682 if (!(ma->ma_valid & MA_LMV)) {
1687 lmv = (struct lmv_mds_md_v1 *)ma->ma_lmm;
1688 if (le32_to_cpu(lmv->lmv_magic) != LMV_MAGIC_V1) {
1690 CERROR("Invalid mds md magic %x "DFID": rc = %d\n",
1691 le32_to_cpu(lmv->lmv_magic), PFID(lu_object_fid(obj)),
1697 tmp_ln_name.ln_name = name;
1698 tmp_ln_name.ln_namelen = namelen;
1701 echo_md_build_name(&tmp_ln_name, info->eti_name, id);
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);
1708 LASSERT(idx < le32_to_cpu(lmv->lmv_stripe_count));
1709 fid_le_to_cpu(&stripe_fid, &lmv->lmv_stripe_fids[idx]);
1711 stripe_obj = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, &stripe_fid,
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);
1720 *new_parent = lu_object_locate(stripe_obj->lo_header, ld->ld_type);
1722 lu_object_put(env, stripe_obj);
1729 static int echo_create_md_object(const struct lu_env *env,
1730 struct echo_device *ed,
1731 struct lu_object *ec_parent,
1733 char *name, int namelen,
1734 __u64 id, __u32 mode, int count,
1735 int stripe_count, int stripe_offset)
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;
1751 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1755 rc = echo_md_dir_stripe_choose(env, ed, parent, name, namelen,
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;
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);
1778 struct lov_user_md_v3 *lum = &info->eti_lum;
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);
1787 spec->sp_cr_flags |= MDS_OPEN_HAS_EA;
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();
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,
1804 /* Create multiple object sequenced by id */
1805 for (i = 0; i < count; i++) {
1806 char *tmp_name = info->eti_name;
1808 echo_md_build_name(lname, tmp_name, id);
1810 rc = echo_md_create_internal(env, ed, lu2md(new_parent),
1811 fid, lname, spec, ma);
1813 CERROR("Can not create child %s: rc = %d\n", tmp_name,
1822 if (new_parent != parent)
1823 lu_object_put(env, new_parent);
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)
1833 struct echo_thread_info *info = echo_env_info(env);
1834 struct lu_fid *fid = &info->eti_fid;
1835 struct lu_object *child;
1839 CDEBUG(D_INFO, "lookup %s in parent "DFID" %p\n", lname->ln_name,
1842 rc = mdo_lookup(env, parent, lname, fid, NULL);
1844 CERROR("lookup %s: rc = %d\n", lname->ln_name, rc);
1845 RETURN(ERR_PTR(rc));
1849 * In the function below, .hs_keycmp resolves to
1850 * lu_obj_hop_keycmp()
1852 /* coverity[overrun-buffer-val] */
1853 child = lu_object_find_at(env, &ed->ed_cl.cd_lu_dev, fid, NULL);
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)
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;
1877 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1881 rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
1886 for (i = 0; i < count; i++) {
1887 struct lu_object *ec_child, *child;
1889 echo_md_build_name(lname, name, id);
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);
1899 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1901 CERROR("Can not locate the child %s\n", lname->ln_name);
1902 lu_object_put(env, ec_child);
1907 CDEBUG(D_RPCTRACE, "Start setattr object "DFID"\n",
1908 PFID(lu_object_fid(child)));
1910 buf->lb_buf = info->eti_xattr_buf;
1911 buf->lb_len = sizeof(info->eti_xattr_buf);
1913 sprintf(name, "%s.test1", XATTR_USER_PREFIX);
1914 rc = mo_xattr_set(env, lu2md(child), buf, name,
1917 CERROR("Can not setattr child "DFID": rc = %d\n",
1918 PFID(lu_object_fid(child)), rc);
1919 lu_object_put(env, ec_child);
1922 CDEBUG(D_RPCTRACE, "End setattr object "DFID"\n",
1923 PFID(lu_object_fid(child)));
1925 lu_object_put(env, ec_child);
1928 if (new_parent != parent)
1929 lu_object_put(env, new_parent);
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)
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;
1953 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
1957 rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
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);
1967 for (i = 0; i < count; i++) {
1968 struct lu_object *ec_child, *child;
1971 echo_md_build_name(lname, name, id);
1972 echo_set_lmm_size(env, ld, ma);
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));
1981 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
1983 CERROR("Can not locate the child %s\n", lname->ln_name);
1984 lu_object_put(env, ec_child);
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);
1992 CERROR("Can not getattr child "DFID": rc = %d\n",
1993 PFID(lu_object_fid(child)), rc);
1994 lu_object_put(env, ec_child);
1997 CDEBUG(D_RPCTRACE, "End getattr object "DFID"\n",
1998 PFID(lu_object_fid(child)));
2000 lu_object_put(env, ec_child);
2003 if (new_parent != parent)
2004 lu_object_put(env, new_parent);
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)
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;
2026 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
2030 rc = echo_md_dir_stripe_choose(env, ed, parent, NULL, 0, id,
2035 /*prepare the requests*/
2036 for (i = 0; i < count; i++) {
2037 echo_md_build_name(lname, name, id);
2039 CDEBUG(D_RPCTRACE, "Start lookup object "DFID" %s %p\n",
2040 PFID(lu_object_fid(new_parent)), lname->ln_name,
2043 rc = mdo_lookup(env, lu2md(new_parent), lname, fid, NULL);
2045 CERROR("Can not lookup child %s: rc = %d\n", name, rc);
2049 CDEBUG(D_RPCTRACE, "End lookup object "DFID" %s %p\n",
2050 PFID(lu_object_fid(new_parent)), lname->ln_name,
2056 if (new_parent != parent)
2057 lu_object_put(env, new_parent);
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,
2068 struct lu_device *ld = ed->ed_next;
2069 struct lu_object *ec_child;
2070 struct lu_object *child;
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,
2079 RETURN(PTR_ERR(ec_child));
2082 child = lu_object_locate(ec_child->lo_header, ld->ld_type);
2084 CERROR("Can not locate the child %s\n", lname->ln_name);
2085 GOTO(out_put, rc = -EINVAL);
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);
2093 CDEBUG(D_RPCTRACE, "Start destroy object "DFID" %s %p\n",
2094 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
2096 rc = mdo_unlink(env, parent, lu2md(child), lname, ma, 0);
2098 CERROR("Can not unlink child %s: rc = %d\n",
2099 lname->ln_name, rc);
2102 CDEBUG(D_RPCTRACE, "End destroy object "DFID" %s %p\n",
2103 PFID(lu_object_fid(&parent->mo_lu)), lname->ln_name, parent);
2105 lu_object_put(env, ec_child);
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,
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;
2126 parent = lu_object_locate(ec_parent->lo_header, ld->ld_type);
2130 rc = echo_md_dir_stripe_choose(env, ed, parent, name, namelen,
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;
2143 lname->ln_name = name;
2144 lname->ln_namelen = namelen;
2145 rc = echo_md_destroy_internal(env, ed, lu2md(new_parent), lname,
2150 /*prepare the requests*/
2151 for (i = 0; i < count; i++) {
2152 char *tmp_name = info->eti_name;
2155 echo_md_build_name(lname, tmp_name, id);
2157 rc = echo_md_destroy_internal(env, ed, lu2md(new_parent), lname,
2160 CERROR("Can not unlink child %s: rc = %d\n", name, rc);
2167 if (new_parent != parent)
2168 lu_object_put(env, new_parent);
2173 static struct lu_object *echo_resolve_path(const struct lu_env *env,
2174 struct echo_device *ed, char *path,
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;
2186 *fid = ed->ed_root_fid;
2189 * In the function below, .hs_keycmp resolves to
2190 * lu_obj_hop_keycmp()
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));
2201 struct lu_object *ld_parent;
2204 e = strsep(&path, "/");
2209 if (!path || path[0] == '\0')
2215 lname->ln_namelen = strlen(e);
2217 ld_parent = lu_object_locate(parent->lo_header, ld->ld_type);
2219 lu_object_put(env, parent);
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)),
2236 RETURN(ERR_PTR(rc));
2241 static void echo_ucred_init(struct lu_env *env)
2243 struct lu_ucred *ucred = lu_ucred(env);
2245 ucred->uc_valid = UCRED_INVALID;
2247 ucred->uc_suppgids[0] = -1;
2248 ucred->uc_suppgids[1] = -1;
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();
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;
2266 static void echo_ucred_fini(struct lu_env *env)
2268 struct lu_ucred *ucred = lu_ucred(env);
2270 ucred->uc_valid = UCRED_INIT;
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)
2277 struct echo_thread_info *info;
2278 struct lu_device *ld = ed->ed_next;
2281 struct lu_object *parent;
2283 int namelen = data->ioc_plen2;
2288 CERROR("MD echo client is not being initialized properly\n");
2292 if (strcmp(ld->ld_type->ldt_name, LUSTRE_MDD_NAME)) {
2293 CERROR("Only support MDD layer right now!\n");
2297 env = cl_env_get(&refcheck);
2299 RETURN(PTR_ERR(env));
2301 rc = lu_env_refill_by_tags(env, ECHO_MD_CTX_TAG, ECHO_SES_TAG);
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;
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,
2317 GOTO(out_free, rc = PTR_ERR(parent));
2321 OBD_ALLOC(name, namelen + 1);
2323 GOTO(out_put, rc = -ENOMEM);
2324 if (copy_from_user(name, data->ioc_pbuf2, namelen))
2325 GOTO(out_name, rc = -EFAULT);
2328 echo_ucred_init(env);
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;
2339 rc = ostid_to_fid(fid, &data->ioc_obdo1.o_oi, 0);
2344 * In the function below, .hs_keycmp resolves to
2345 * lu_obj_hop_keycmp()
2347 /* coverity[overrun-buffer-val] */
2348 rc = echo_create_md_object(env, ed, parent, fid, name, namelen,
2349 id, mode, count, stripe_count,
2353 case ECHO_MD_DESTROY:
2354 case ECHO_MD_RMDIR: {
2355 __u32 mode = data->ioc_obdo2.o_mode;
2357 rc = echo_destroy_object(env, ed, parent, name, namelen,
2361 case ECHO_MD_LOOKUP:
2362 rc = echo_lookup_object(env, ed, parent, id, count);
2364 case ECHO_MD_GETATTR:
2365 rc = echo_getattr_object(env, ed, parent, id, count);
2367 case ECHO_MD_SETATTR:
2368 rc = echo_setattr_object(env, ed, parent, id, count);
2371 CERROR("unknown command %d\n", command);
2375 echo_ucred_fini(env);
2379 OBD_FREE(name, namelen + 1);
2381 lu_object_put(env, parent);
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;
2388 cl_env_put(env, &refcheck);
2391 #endif /* HAVE_SERVER_SUPPORT */
2393 static int echo_create_object(const struct lu_env *env, struct echo_device *ed,
2396 struct echo_object *eco;
2397 struct echo_client_obd *ec = ed->ed_ec;
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));
2409 if (ostid_id(&oa->o_oi) == 0) {
2410 rc = ostid_set_id(&oa->o_oi, ++last_object_id);
2415 rc = obd_create(env, ec->ec_exp, oa);
2417 CERROR("Cannot create objects: rc = %d\n", rc);
2423 oa->o_valid |= OBD_MD_FLID;
2425 eco = cl_echo_object_find(ed, &oa->o_oi);
2427 GOTO(failed, rc = PTR_ERR(eco));
2428 cl_echo_object_put(eco);
2430 CDEBUG(D_INFO, "oa oid "DOSTID"\n", POSTID(&oa->o_oi));
2434 if (created && rc != 0)
2435 obd_destroy(env, ec->ec_exp, oa);
2438 CERROR("create object failed with: rc = %d\n", rc);
2443 static int echo_get_object(struct echo_object **ecop, struct echo_device *ed,
2446 struct echo_object *eco;
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));
2458 eco = cl_echo_object_find(ed, &oa->o_oi);
2467 static void echo_put_object(struct echo_object *eco)
2471 rc = cl_echo_object_put(eco);
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);
2477 static void echo_client_page_debug_setup(struct page *page, int rw, u64 id,
2478 u64 offset, u64 count)
2485 /* no partial pages on the client */
2486 LASSERT(count == PAGE_SIZE);
2490 for (delta = 0; delta < PAGE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2491 if (rw == OBD_BRW_WRITE) {
2492 stripe_off = offset + delta;
2495 stripe_off = 0xdeadbeef00c0ffeeULL;
2496 stripe_id = 0xdeadbeef00c0ffeeULL;
2498 block_debug_setup(addr + delta, OBD_ECHO_BLOCK_SIZE,
2499 stripe_off, stripe_id);
2506 echo_client_page_debug_check(struct page *page, u64 id, u64 offset, u64 count)
2515 /* no partial pages on the client */
2516 LASSERT(count == PAGE_SIZE);
2520 for (rc = delta = 0; delta < PAGE_SIZE; delta += OBD_ECHO_BLOCK_SIZE) {
2521 stripe_off = offset + delta;
2524 rc2 = block_debug_check("test_brw",
2525 addr + delta, OBD_ECHO_BLOCK_SIZE,
2526 stripe_off, stripe_id);
2528 CERROR("Error in echo object %#llx\n", id);
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)
2542 struct brw_page *pga;
2543 struct brw_page *pgp;
2544 struct page **pages;
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);
2557 gfp_mask = ((ostid_id(&oa->o_oi) & 2) == 0) ? GFP_KERNEL : GFP_HIGHUSER;
2559 LASSERT(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ);
2561 if ((count & (~PAGE_MASK)) != 0)
2564 /* XXX think again with misaligned I/O */
2565 npages = count >> PAGE_SHIFT;
2567 if (rw == OBD_BRW_WRITE)
2568 brw_flags = OBD_BRW_ASYNC;
2570 OBD_ALLOC(pga, npages * sizeof(*pga));
2574 OBD_ALLOC(pages, npages * sizeof(*pages));
2576 OBD_FREE(pga, npages * sizeof(*pga));
2580 for (i = 0, pgp = pga, off = offset;
2582 i++, pgp++, off += PAGE_SIZE) {
2584 LASSERT(pgp->pg == NULL); /* for cleanup */
2587 pgp->pg = alloc_page(gfp_mask);
2592 pgp->count = PAGE_SIZE;
2594 pgp->flag = brw_flags;
2597 echo_client_page_debug_setup(pgp->pg, rw,
2598 ostid_id(&oa->o_oi), off,
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);
2607 if (rc != 0 || rw != OBD_BRW_READ)
2610 for (i = 0, pgp = pga; i < npages; i++, pgp++) {
2617 vrc = echo_client_page_debug_check(pgp->pg,
2618 ostid_id(&oa->o_oi),
2621 if (vrc != 0 && rc == 0)
2624 __free_page(pgp->pg);
2626 OBD_FREE(pga, npages * sizeof(*pga));
2627 OBD_FREE(pages, npages * sizeof(*pages));
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)
2637 struct obd_ioobj ioo;
2638 struct niobuf_local *lnb;
2639 struct niobuf_remote rnb;
2641 u64 npages, tot_pages, apc;
2642 int i, ret = 0, brw_flags = 0;
2645 if (count <= 0 || (count & ~PAGE_MASK) != 0)
2648 apc = npages = batch >> PAGE_SHIFT;
2649 tot_pages = count >> PAGE_SHIFT;
2651 OBD_ALLOC_LARGE(lnb, apc * sizeof(*lnb));
2655 if (rw == OBD_BRW_WRITE && async)
2656 brw_flags |= OBD_BRW_ASYNC;
2658 obdo_to_ioobj(oa, &ioo);
2662 for (; tot_pages > 0; tot_pages -= npages) {
2665 if (tot_pages < npages)
2668 rnb.rnb_offset = off;
2669 rnb.rnb_len = npages * PAGE_SIZE;
2670 rnb.rnb_flags = brw_flags;
2672 off += npages * PAGE_SIZE;
2675 ret = obd_preprw(env, rw, exp, oa, 1, &ioo, &rnb, &lpages, lnb);
2679 for (i = 0; i < lpages; i++) {
2680 struct page *page = lnb[i].lnb_page;
2682 /* read past eof? */
2683 if (!page && lnb[i].lnb_rc == 0)
2687 lnb[i].lnb_flags |= OBD_BRW_ASYNC;
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)
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,
2700 echo_client_page_debug_check(page,
2701 ostid_id(&oa->o_oi),
2702 lnb[i].lnb_file_offset,
2706 ret = obd_commitrw(env, rw, exp, oa, 1, &ioo, &rnb, npages, lnb,
2711 /* Reuse env context. */
2712 lu_context_exit((struct lu_context *)&env->le_ctx);
2713 lu_context_enter((struct lu_context *)&env->le_ctx);
2717 OBD_FREE_LARGE(lnb, apc * sizeof(*lnb));
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)
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;
2736 LASSERT(oa->o_valid & OBD_MD_FLGROUP);
2738 rc = echo_get_object(&eco, ed, oa);
2742 oa->o_valid &= ~OBD_MD_FLHANDLE;
2744 /* OFD/obdfilter works only via prep/commit */
2745 test_mode = (long)data->ioc_pbuf1;
2746 if (!ed->ed_next && test_mode != 3) {
2748 data->ioc_plen1 = data->ioc_count;
2754 /* Truncate batch size to maximum */
2755 if (data->ioc_plen1 > PTLRPC_MAX_BRW_SIZE)
2756 data->ioc_plen1 = PTLRPC_MAX_BRW_SIZE;
2758 switch (test_mode) {
2762 rc = echo_client_kbrw(ed, rw, oa, eco, data->ioc_offset,
2763 data->ioc_count, async);
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);
2774 echo_put_object(eco);
2780 echo_client_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2781 void *karg, void __user *uarg)
2783 #ifdef HAVE_SERVER_SUPPORT
2784 struct tgt_session_info *tsi;
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;
2792 unsigned long env_tags = 0;
2796 int rw = OBD_BRW_READ;
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);
2806 /* This FID is unpacked just for validation at this point */
2807 rc = ostid_to_fid(&fid, &oa->o_oi, 0);
2811 env = cl_env_get(&refcheck);
2813 RETURN(PTR_ERR(env));
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;
2822 env_tags = ECHO_DT_CTX_TAG;
2824 rc = lu_env_refill_by_tags(env, env_tags, ECHO_SES_TAG);
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;
2836 case OBD_IOC_CREATE: /* may create echo object */
2837 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2838 GOTO(out, rc = -EPERM);
2840 rc = echo_create_object(env, ed, oa);
2843 #ifdef HAVE_SERVER_SUPPORT
2844 case OBD_IOC_ECHO_MD: {
2851 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2852 GOTO(out, rc = -EPERM);
2854 count = data->ioc_count;
2855 cmd = data->ioc_command;
2857 id = data->ioc_obdo2.o_oi.oi.oi_id;
2858 dirlen = data->ioc_plen1;
2859 OBD_ALLOC(dir, dirlen + 1);
2861 GOTO(out, rc = -ENOMEM);
2863 if (copy_from_user(dir, data->ioc_pbuf1, dirlen)) {
2864 OBD_FREE(dir, data->ioc_plen1 + 1);
2865 GOTO(out, rc = -EFAULT);
2868 rc = echo_md_handler(ed, cmd, dir, dirlen, id, count, data);
2869 OBD_FREE(dir, dirlen + 1);
2872 case OBD_IOC_ECHO_ALLOC_SEQ: {
2876 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2877 GOTO(out, rc = -EPERM);
2879 rc = seq_client_get_seq(env, ed->ed_cl_seq, &seq);
2881 CERROR("%s: Can not alloc seq: rc = %d\n",
2886 if (copy_to_user(data->ioc_pbuf1, &seq, data->ioc_plen1))
2889 max_count = LUSTRE_METADATA_SEQ_MAX_WIDTH;
2890 if (copy_to_user(data->ioc_pbuf2, &max_count,
2895 #endif /* HAVE_SERVER_SUPPORT */
2896 case OBD_IOC_DESTROY:
2897 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2898 GOTO(out, rc = -EPERM);
2900 rc = echo_get_object(&eco, ed, oa);
2902 rc = obd_destroy(env, ec->ec_exp, oa);
2904 eco->eo_deleted = 1;
2905 echo_put_object(eco);
2909 case OBD_IOC_GETATTR:
2910 rc = echo_get_object(&eco, ed, oa);
2912 rc = obd_getattr(env, ec->ec_exp, oa);
2913 echo_put_object(eco);
2917 case OBD_IOC_SETATTR:
2918 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2919 GOTO(out, rc = -EPERM);
2921 rc = echo_get_object(&eco, ed, oa);
2923 rc = obd_setattr(env, ec->ec_exp, oa);
2924 echo_put_object(eco);
2928 case OBD_IOC_BRW_WRITE:
2929 if (!cfs_capable(CFS_CAP_SYS_ADMIN))
2930 GOTO(out, rc = -EPERM);
2934 case OBD_IOC_BRW_READ:
2935 rc = echo_client_brw_ioctl(env, rw, exp, data);
2939 CERROR("echo_ioctl(): unrecognised ioctl %#x\n", cmd);
2940 GOTO(out, rc = -ENOTTY);
2946 cl_env_put(env, &refcheck);
2951 static int echo_client_setup(const struct lu_env *env,
2952 struct obd_device *obddev, struct lustre_cfg *lcfg)
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;
2961 if (lcfg->lcfg_bufcount < 2 || LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
2962 CERROR("requires a TARGET OBD name\n");
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));
2973 spin_lock_init(&ec->ec_lock);
2974 INIT_LIST_HEAD(&ec->ec_objects);
2975 INIT_LIST_HEAD(&ec->ec_locks);
2978 lu_context_tags_update(ECHO_DT_CTX_TAG);
2979 lu_session_tags_update(ECHO_SES_TAG);
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);
2986 "Local operations are NOT supported on client side. Only remote operations are supported. Metadata client must be run on server side.\n");
2991 OBD_ALLOC(ocd, sizeof(*ocd));
2993 CERROR("Can't alloc ocd connecting to %s\n",
2994 lustre_cfg_string(lcfg, 1));
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 |
3003 ocd->ocd_brw_size = DT_MAX_BRW_SIZE;
3004 ocd->ocd_version = LUSTRE_VERSION_CODE;
3005 ocd->ocd_group = FID_SEQ_ECHO;
3007 rc = obd_connect(env, &ec->ec_exp, tgt, &echo_uuid, ocd, NULL);
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);
3015 OBD_FREE(ocd, sizeof(*ocd));
3018 CERROR("fail to connect to device %s\n",
3019 lustre_cfg_string(lcfg, 1));
3026 static int echo_client_cleanup(struct obd_device *obddev)
3028 struct echo_device *ed = obd2echo_dev(obddev);
3029 struct echo_client_obd *ec = &obddev->u.echo_client;
3033 /*Do nothing for Metadata echo client*/
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);
3044 "This is client-side only module, does not support metadata echo client.\n");
3049 if (!list_empty(&obddev->obd_exports)) {
3050 CERROR("still has clients!\n");
3054 LASSERT(refcount_read(&ec->ec_exp->exp_handle.h_ref) > 0);
3055 rc = obd_disconnect(ec->ec_exp);
3057 CERROR("fail to disconnect device: %d\n", rc);
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)
3068 struct lustre_handle conn = { 0 };
3071 rc = class_connect(&conn, src, cluuid);
3073 *exp = class_conn2export(&conn);
3078 static int echo_client_disconnect(struct obd_export *exp)
3084 GOTO(out, rc = -EINVAL);
3086 rc = class_disconnect(exp);
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
3099 static int __init obdecho_init(void)
3104 LCONSOLE_INFO("Echo OBD driver; http://www.lustre.org/\n");
3106 LASSERT(PAGE_SIZE % OBD_ECHO_BLOCK_SIZE == 0);
3108 # ifdef HAVE_SERVER_SUPPORT
3109 rc = echo_persistent_pages_init();
3113 rc = class_register_type(&echo_obd_ops, NULL, true, NULL,
3114 LUSTRE_ECHO_NAME, &echo_srv_type);
3119 rc = lu_kmem_init(echo_caches);
3121 rc = class_register_type(&echo_client_obd_ops, NULL, false,
3122 NULL, LUSTRE_ECHO_CLIENT_NAME,
3125 lu_kmem_fini(echo_caches);
3128 # ifdef HAVE_SERVER_SUPPORT
3132 class_unregister_type(LUSTRE_ECHO_NAME);
3134 echo_persistent_pages_fini();
3140 static void __exit obdecho_exit(void)
3142 class_unregister_type(LUSTRE_ECHO_CLIENT_NAME);
3143 lu_kmem_fini(echo_caches);
3145 #ifdef HAVE_SERVER_SUPPORT
3146 class_unregister_type(LUSTRE_ECHO_NAME);
3147 echo_persistent_pages_fini();
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");
3156 module_init(obdecho_init);
3157 module_exit(obdecho_exit);
3159 /** @} echo_client */