1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
6 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
8 * This program is free software; you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License version 2 only,
10 * as published by the Free Software Foundation.
12 * This program is distributed in the hope that it will be useful, but
13 * WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * General Public License version 2 for more details (a copy is included
16 * in the LICENSE file that accompanied this code).
18 * You should have received a copy of the GNU General Public License
19 * version 2 along with this program; If not, see
20 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
22 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23 * CA 95054 USA or visit www.sun.com if you need additional information or
29 * Copyright 2008 Sun Microsystems, Inc. All rights reserved
30 * Use is subject to license terms.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
38 # define EXPORT_SYMTAB
40 #define DEBUG_SUBSYSTEM S_OSC
42 #include <libcfs/libcfs.h>
45 # include <liblustre.h>
48 #include <lustre_dlm.h>
49 #include <lustre_net.h>
50 #include <lustre/lustre_user.h>
51 #include <obd_cksum.h>
59 #include <lustre_ha.h>
60 #include <lprocfs_status.h>
61 #include <lustre_log.h>
62 #include <lustre_debug.h>
63 #include <lustre_param.h>
64 #include "osc_internal.h"
66 static quota_interface_t *quota_interface = NULL;
67 extern quota_interface_t osc_quota_interface;
69 static void osc_release_ppga(struct brw_page **ppga, obd_count count);
70 static int brw_interpret(const struct lu_env *env,
71 struct ptlrpc_request *req, void *data, int rc);
72 int osc_cleanup(struct obd_device *obd);
74 /* Pack OSC object metadata for disk storage (LE byte order). */
75 static int osc_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
76 struct lov_stripe_md *lsm)
81 lmm_size = sizeof(**lmmp);
86 OBD_FREE(*lmmp, lmm_size);
92 OBD_ALLOC(*lmmp, lmm_size);
98 LASSERT(lsm->lsm_object_id);
99 LASSERT_MDS_GROUP(lsm->lsm_object_gr);
100 (*lmmp)->lmm_object_id = cpu_to_le64(lsm->lsm_object_id);
101 (*lmmp)->lmm_object_gr = cpu_to_le64(lsm->lsm_object_gr);
107 /* Unpack OSC object metadata from disk storage (LE byte order). */
108 static int osc_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
109 struct lov_mds_md *lmm, int lmm_bytes)
115 if (lmm_bytes < sizeof (*lmm)) {
116 CERROR("lov_mds_md too small: %d, need %d\n",
117 lmm_bytes, (int)sizeof(*lmm));
120 /* XXX LOV_MAGIC etc check? */
122 if (lmm->lmm_object_id == 0) {
123 CERROR("lov_mds_md: zero lmm_object_id\n");
128 lsm_size = lov_stripe_md_size(1);
132 if (*lsmp != NULL && lmm == NULL) {
133 OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
134 OBD_FREE(*lsmp, lsm_size);
140 OBD_ALLOC(*lsmp, lsm_size);
143 OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
144 if ((*lsmp)->lsm_oinfo[0] == NULL) {
145 OBD_FREE(*lsmp, lsm_size);
148 loi_init((*lsmp)->lsm_oinfo[0]);
152 /* XXX zero *lsmp? */
153 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
154 (*lsmp)->lsm_object_gr = le64_to_cpu (lmm->lmm_object_gr);
155 LASSERT((*lsmp)->lsm_object_id);
156 LASSERT_MDS_GROUP((*lsmp)->lsm_object_gr);
159 (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
164 static inline void osc_pack_capa(struct ptlrpc_request *req,
165 struct ost_body *body, void *capa)
167 struct obd_capa *oc = (struct obd_capa *)capa;
168 struct lustre_capa *c;
173 c = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
176 body->oa.o_valid |= OBD_MD_FLOSSCAPA;
177 DEBUG_CAPA(D_SEC, c, "pack");
180 static inline void osc_pack_req_body(struct ptlrpc_request *req,
181 struct obd_info *oinfo)
183 struct ost_body *body;
185 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
188 body->oa = *oinfo->oi_oa;
189 osc_pack_capa(req, body, oinfo->oi_capa);
192 static inline void osc_set_capa_size(struct ptlrpc_request *req,
193 const struct req_msg_field *field,
197 req_capsule_set_size(&req->rq_pill, field, RCL_CLIENT, 0);
199 /* it is already calculated as sizeof struct obd_capa */
203 static int osc_getattr_interpret(const struct lu_env *env,
204 struct ptlrpc_request *req,
205 struct osc_async_args *aa, int rc)
207 struct ost_body *body;
213 body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
214 lustre_swab_ost_body);
216 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
217 memcpy(aa->aa_oi->oi_oa, &body->oa, sizeof(*aa->aa_oi->oi_oa));
219 /* This should really be sent by the OST */
220 aa->aa_oi->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
221 aa->aa_oi->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
223 CDEBUG(D_INFO, "can't unpack ost_body\n");
225 aa->aa_oi->oi_oa->o_valid = 0;
228 rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
232 static int osc_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
233 struct ptlrpc_request_set *set)
235 struct ptlrpc_request *req;
236 struct osc_async_args *aa;
240 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
244 osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
245 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
247 ptlrpc_request_free(req);
251 osc_pack_req_body(req, oinfo);
253 ptlrpc_request_set_replen(req);
254 req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_getattr_interpret;
256 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
257 aa = ptlrpc_req_async_args(req);
260 ptlrpc_set_add_req(set, req);
264 static int osc_getattr(struct obd_export *exp, struct obd_info *oinfo)
266 struct ptlrpc_request *req;
267 struct ost_body *body;
271 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
275 osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
276 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
278 ptlrpc_request_free(req);
282 osc_pack_req_body(req, oinfo);
284 ptlrpc_request_set_replen(req);
286 rc = ptlrpc_queue_wait(req);
290 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
292 GOTO(out, rc = -EPROTO);
294 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
295 *oinfo->oi_oa = body->oa;
297 /* This should really be sent by the OST */
298 oinfo->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
299 oinfo->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
303 ptlrpc_req_finished(req);
307 static int osc_setattr(struct obd_export *exp, struct obd_info *oinfo,
308 struct obd_trans_info *oti)
310 struct ptlrpc_request *req;
311 struct ost_body *body;
315 LASSERTF(!(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP) ||
316 CHECK_MDS_GROUP(oinfo->oi_oa->o_gr),
317 "oinfo->oi_oa->o_valid="LPU64" oinfo->oi_oa->o_gr="LPU64"\n",
318 oinfo->oi_oa->o_valid, oinfo->oi_oa->o_gr);
320 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
324 osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
325 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
327 ptlrpc_request_free(req);
331 osc_pack_req_body(req, oinfo);
333 ptlrpc_request_set_replen(req);
335 rc = ptlrpc_queue_wait(req);
339 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
341 GOTO(out, rc = -EPROTO);
343 *oinfo->oi_oa = body->oa;
347 ptlrpc_req_finished(req);
351 static int osc_setattr_interpret(const struct lu_env *env,
352 struct ptlrpc_request *req,
353 struct osc_async_args *aa, int rc)
355 struct ost_body *body;
361 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
363 GOTO(out, rc = -EPROTO);
365 *aa->aa_oi->oi_oa = body->oa;
367 rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
371 static int osc_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
372 struct obd_trans_info *oti,
373 struct ptlrpc_request_set *rqset)
375 struct ptlrpc_request *req;
376 struct osc_async_args *aa;
380 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
384 osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
385 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
387 ptlrpc_request_free(req);
391 osc_pack_req_body(req, oinfo);
393 ptlrpc_request_set_replen(req);
395 if (oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE) {
397 oinfo->oi_oa->o_lcookie = *oti->oti_logcookies;
400 /* do mds to ost setattr asynchronously */
402 /* Do not wait for response. */
403 ptlrpcd_add_req(req, PSCOPE_OTHER);
405 req->rq_interpret_reply =
406 (ptlrpc_interpterer_t)osc_setattr_interpret;
408 CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
409 aa = ptlrpc_req_async_args(req);
412 ptlrpc_set_add_req(rqset, req);
418 int osc_real_create(struct obd_export *exp, struct obdo *oa,
419 struct lov_stripe_md **ea, struct obd_trans_info *oti)
421 struct ptlrpc_request *req;
422 struct ost_body *body;
423 struct lov_stripe_md *lsm;
432 rc = obd_alloc_memmd(exp, &lsm);
437 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
439 GOTO(out, rc = -ENOMEM);
441 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
443 ptlrpc_request_free(req);
447 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
451 ptlrpc_request_set_replen(req);
453 if ((oa->o_valid & OBD_MD_FLFLAGS) &&
454 oa->o_flags == OBD_FL_DELORPHAN) {
456 "delorphan from OST integration");
457 /* Don't resend the delorphan req */
458 req->rq_no_resend = req->rq_no_delay = 1;
461 rc = ptlrpc_queue_wait(req);
465 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
467 GOTO(out_req, rc = -EPROTO);
471 /* This should really be sent by the OST */
472 oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
473 oa->o_valid |= OBD_MD_FLBLKSZ;
475 /* XXX LOV STACKING: the lsm that is passed to us from LOV does not
476 * have valid lsm_oinfo data structs, so don't go touching that.
477 * This needs to be fixed in a big way.
479 lsm->lsm_object_id = oa->o_id;
480 lsm->lsm_object_gr = oa->o_gr;
484 oti->oti_transno = lustre_msg_get_transno(req->rq_repmsg);
486 if (oa->o_valid & OBD_MD_FLCOOKIE) {
487 if (!oti->oti_logcookies)
488 oti_alloc_cookies(oti, 1);
489 *oti->oti_logcookies = oa->o_lcookie;
493 CDEBUG(D_HA, "transno: "LPD64"\n",
494 lustre_msg_get_transno(req->rq_repmsg));
496 ptlrpc_req_finished(req);
499 obd_free_memmd(exp, &lsm);
503 static int osc_punch_interpret(const struct lu_env *env,
504 struct ptlrpc_request *req,
505 struct osc_punch_args *aa, int rc)
507 struct ost_body *body;
513 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
515 GOTO(out, rc = -EPROTO);
517 *aa->pa_oa = body->oa;
519 rc = aa->pa_upcall(aa->pa_cookie, rc);
523 int osc_punch_base(struct obd_export *exp, struct obdo *oa,
524 struct obd_capa *capa,
525 obd_enqueue_update_f upcall, void *cookie,
526 struct ptlrpc_request_set *rqset)
528 struct ptlrpc_request *req;
529 struct osc_punch_args *aa;
530 struct ost_body *body;
534 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_PUNCH);
538 osc_set_capa_size(req, &RMF_CAPA1, capa);
539 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
541 ptlrpc_request_free(req);
544 req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
545 ptlrpc_at_set_req_timeout(req);
547 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
550 osc_pack_capa(req, body, capa);
552 ptlrpc_request_set_replen(req);
555 req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_punch_interpret;
556 CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
557 aa = ptlrpc_req_async_args(req);
559 aa->pa_upcall = upcall;
560 aa->pa_cookie = cookie;
561 if (rqset == PTLRPCD_SET)
562 ptlrpcd_add_req(req, PSCOPE_OTHER);
564 ptlrpc_set_add_req(rqset, req);
569 static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
570 struct obd_trans_info *oti,
571 struct ptlrpc_request_set *rqset)
573 oinfo->oi_oa->o_size = oinfo->oi_policy.l_extent.start;
574 oinfo->oi_oa->o_blocks = oinfo->oi_policy.l_extent.end;
575 oinfo->oi_oa->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
576 return osc_punch_base(exp, oinfo->oi_oa, oinfo->oi_capa,
577 oinfo->oi_cb_up, oinfo, rqset);
580 static int osc_sync(struct obd_export *exp, struct obdo *oa,
581 struct lov_stripe_md *md, obd_size start, obd_size end,
584 struct ptlrpc_request *req;
585 struct ost_body *body;
590 CDEBUG(D_INFO, "oa NULL\n");
594 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
598 osc_set_capa_size(req, &RMF_CAPA1, capa);
599 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
601 ptlrpc_request_free(req);
605 /* overload the size and blocks fields in the oa with start/end */
606 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
609 body->oa.o_size = start;
610 body->oa.o_blocks = end;
611 body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
612 osc_pack_capa(req, body, capa);
614 ptlrpc_request_set_replen(req);
616 rc = ptlrpc_queue_wait(req);
620 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
622 GOTO(out, rc = -EPROTO);
628 ptlrpc_req_finished(req);
632 /* Find and cancel locally locks matched by @mode in the resource found by
633 * @objid. Found locks are added into @cancel list. Returns the amount of
634 * locks added to @cancels list. */
635 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
636 struct list_head *cancels, ldlm_mode_t mode,
639 struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
640 struct ldlm_res_id res_id;
641 struct ldlm_resource *res;
645 osc_build_res_name(oa->o_id, oa->o_gr, &res_id);
646 res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
650 LDLM_RESOURCE_ADDREF(res);
651 count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
652 lock_flags, 0, NULL);
653 LDLM_RESOURCE_DELREF(res);
654 ldlm_resource_putref(res);
658 static int osc_destroy_interpret(const struct lu_env *env,
659 struct ptlrpc_request *req, void *data,
662 struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
664 atomic_dec(&cli->cl_destroy_in_flight);
665 cfs_waitq_signal(&cli->cl_destroy_waitq);
669 static int osc_can_send_destroy(struct client_obd *cli)
671 if (atomic_inc_return(&cli->cl_destroy_in_flight) <=
672 cli->cl_max_rpcs_in_flight) {
673 /* The destroy request can be sent */
676 if (atomic_dec_return(&cli->cl_destroy_in_flight) <
677 cli->cl_max_rpcs_in_flight) {
679 * The counter has been modified between the two atomic
682 cfs_waitq_signal(&cli->cl_destroy_waitq);
687 /* Destroy requests can be async always on the client, and we don't even really
688 * care about the return code since the client cannot do anything at all about
690 * When the MDS is unlinking a filename, it saves the file objects into a
691 * recovery llog, and these object records are cancelled when the OST reports
692 * they were destroyed and sync'd to disk (i.e. transaction committed).
693 * If the client dies, or the OST is down when the object should be destroyed,
694 * the records are not cancelled, and when the OST reconnects to the MDS next,
695 * it will retrieve the llog unlink logs and then sends the log cancellation
696 * cookies to the MDS after committing destroy transactions. */
697 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
698 struct lov_stripe_md *ea, struct obd_trans_info *oti,
699 struct obd_export *md_export, void *capa)
701 struct client_obd *cli = &exp->exp_obd->u.cli;
702 struct ptlrpc_request *req;
703 struct ost_body *body;
704 CFS_LIST_HEAD(cancels);
709 CDEBUG(D_INFO, "oa NULL\n");
713 count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
714 LDLM_FL_DISCARD_DATA);
716 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
718 ldlm_lock_list_put(&cancels, l_bl_ast, count);
722 osc_set_capa_size(req, &RMF_CAPA1, (struct obd_capa *)capa);
723 rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
726 ptlrpc_request_free(req);
730 req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
731 ptlrpc_at_set_req_timeout(req);
733 if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE)
734 oa->o_lcookie = *oti->oti_logcookies;
735 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
739 osc_pack_capa(req, body, (struct obd_capa *)capa);
740 ptlrpc_request_set_replen(req);
742 /* don't throttle destroy RPCs for the MDT */
743 if (!(cli->cl_import->imp_connect_flags_orig & OBD_CONNECT_MDS)) {
744 req->rq_interpret_reply = osc_destroy_interpret;
745 if (!osc_can_send_destroy(cli)) {
746 struct l_wait_info lwi = { 0 };
749 * Wait until the number of on-going destroy RPCs drops
750 * under max_rpc_in_flight
752 l_wait_event_exclusive(cli->cl_destroy_waitq,
753 osc_can_send_destroy(cli), &lwi);
757 /* Do not wait for response */
758 ptlrpcd_add_req(req, PSCOPE_OTHER);
762 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
765 obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
767 LASSERT(!(oa->o_valid & bits));
770 client_obd_list_lock(&cli->cl_loi_list_lock);
771 oa->o_dirty = cli->cl_dirty;
772 if (cli->cl_dirty - cli->cl_dirty_transit > cli->cl_dirty_max) {
773 CERROR("dirty %lu - %lu > dirty_max %lu\n",
774 cli->cl_dirty, cli->cl_dirty_transit, cli->cl_dirty_max);
776 } else if (atomic_read(&obd_dirty_pages) -
777 atomic_read(&obd_dirty_transit_pages) > obd_max_dirty_pages){
778 CERROR("dirty %d - %d > system dirty_max %d\n",
779 atomic_read(&obd_dirty_pages),
780 atomic_read(&obd_dirty_transit_pages),
781 obd_max_dirty_pages);
783 } else if (cli->cl_dirty_max - cli->cl_dirty > 0x7fffffff) {
784 CERROR("dirty %lu - dirty_max %lu too big???\n",
785 cli->cl_dirty, cli->cl_dirty_max);
788 long max_in_flight = (cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT)*
789 (cli->cl_max_rpcs_in_flight + 1);
790 oa->o_undirty = max(cli->cl_dirty_max, max_in_flight);
792 oa->o_grant = cli->cl_avail_grant;
793 oa->o_dropped = cli->cl_lost_grant;
794 cli->cl_lost_grant = 0;
795 client_obd_list_unlock(&cli->cl_loi_list_lock);
796 CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
797 oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
801 static void osc_update_next_shrink(struct client_obd *cli)
803 int time = GRANT_SHRINK_INTERVAL;
804 cli->cl_next_shrink_grant = cfs_time_shift(time);
805 CDEBUG(D_CACHE, "next time %ld to shrink grant \n",
806 cli->cl_next_shrink_grant);
809 /* caller must hold loi_list_lock */
810 static void osc_consume_write_grant(struct client_obd *cli,
811 struct brw_page *pga)
813 LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock);
814 LASSERT(!(pga->flag & OBD_BRW_FROM_GRANT));
815 atomic_inc(&obd_dirty_pages);
816 cli->cl_dirty += CFS_PAGE_SIZE;
817 cli->cl_avail_grant -= CFS_PAGE_SIZE;
818 pga->flag |= OBD_BRW_FROM_GRANT;
819 CDEBUG(D_CACHE, "using %lu grant credits for brw %p page %p\n",
820 CFS_PAGE_SIZE, pga, pga->pg);
821 LASSERT(cli->cl_avail_grant >= 0);
822 osc_update_next_shrink(cli);
825 /* the companion to osc_consume_write_grant, called when a brw has completed.
826 * must be called with the loi lock held. */
827 static void osc_release_write_grant(struct client_obd *cli,
828 struct brw_page *pga, int sent)
830 int blocksize = cli->cl_import->imp_obd->obd_osfs.os_bsize ? : 4096;
833 LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock);
834 if (!(pga->flag & OBD_BRW_FROM_GRANT)) {
839 pga->flag &= ~OBD_BRW_FROM_GRANT;
840 atomic_dec(&obd_dirty_pages);
841 cli->cl_dirty -= CFS_PAGE_SIZE;
842 if (pga->flag & OBD_BRW_NOCACHE) {
843 pga->flag &= ~OBD_BRW_NOCACHE;
844 atomic_dec(&obd_dirty_transit_pages);
845 cli->cl_dirty_transit -= CFS_PAGE_SIZE;
848 cli->cl_lost_grant += CFS_PAGE_SIZE;
849 CDEBUG(D_CACHE, "lost grant: %lu avail grant: %lu dirty: %lu\n",
850 cli->cl_lost_grant, cli->cl_avail_grant, cli->cl_dirty);
851 } else if (CFS_PAGE_SIZE != blocksize && pga->count != CFS_PAGE_SIZE) {
852 /* For short writes we shouldn't count parts of pages that
853 * span a whole block on the OST side, or our accounting goes
854 * wrong. Should match the code in filter_grant_check. */
855 int offset = pga->off & ~CFS_PAGE_MASK;
856 int count = pga->count + (offset & (blocksize - 1));
857 int end = (offset + pga->count) & (blocksize - 1);
859 count += blocksize - end;
861 cli->cl_lost_grant += CFS_PAGE_SIZE - count;
862 CDEBUG(D_CACHE, "lost %lu grant: %lu avail: %lu dirty: %lu\n",
863 CFS_PAGE_SIZE - count, cli->cl_lost_grant,
864 cli->cl_avail_grant, cli->cl_dirty);
870 static unsigned long rpcs_in_flight(struct client_obd *cli)
872 return cli->cl_r_in_flight + cli->cl_w_in_flight;
875 /* caller must hold loi_list_lock */
876 void osc_wake_cache_waiters(struct client_obd *cli)
878 struct list_head *l, *tmp;
879 struct osc_cache_waiter *ocw;
882 list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
883 /* if we can't dirty more, we must wait until some is written */
884 if ((cli->cl_dirty + CFS_PAGE_SIZE > cli->cl_dirty_max) ||
885 (atomic_read(&obd_dirty_pages) + 1 > obd_max_dirty_pages)) {
886 CDEBUG(D_CACHE, "no dirty room: dirty: %ld "
887 "osc max %ld, sys max %d\n", cli->cl_dirty,
888 cli->cl_dirty_max, obd_max_dirty_pages);
892 /* if still dirty cache but no grant wait for pending RPCs that
893 * may yet return us some grant before doing sync writes */
894 if (cli->cl_w_in_flight && cli->cl_avail_grant < CFS_PAGE_SIZE) {
895 CDEBUG(D_CACHE, "%u BRW writes in flight, no grant\n",
896 cli->cl_w_in_flight);
900 ocw = list_entry(l, struct osc_cache_waiter, ocw_entry);
901 list_del_init(&ocw->ocw_entry);
902 if (cli->cl_avail_grant < CFS_PAGE_SIZE) {
903 /* no more RPCs in flight to return grant, do sync IO */
904 ocw->ocw_rc = -EDQUOT;
905 CDEBUG(D_INODE, "wake oap %p for sync\n", ocw->ocw_oap);
907 osc_consume_write_grant(cli,
908 &ocw->ocw_oap->oap_brw_page);
911 cfs_waitq_signal(&ocw->ocw_waitq);
917 static void __osc_update_grant(struct client_obd *cli, obd_size grant)
919 client_obd_list_lock(&cli->cl_loi_list_lock);
920 cli->cl_avail_grant += grant;
921 client_obd_list_unlock(&cli->cl_loi_list_lock);
924 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
926 if (body->oa.o_valid & OBD_MD_FLGRANT) {
927 CDEBUG(D_CACHE, "got "LPU64" extra grant\n", body->oa.o_grant);
928 __osc_update_grant(cli, body->oa.o_grant);
932 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
933 void *key, obd_count vallen, void *val,
934 struct ptlrpc_request_set *set);
936 static int osc_shrink_grant_interpret(const struct lu_env *env,
937 struct ptlrpc_request *req,
940 struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
941 struct obdo *oa = ((struct osc_grant_args *)aa)->aa_oa;
942 struct ost_body *body;
945 __osc_update_grant(cli, oa->o_grant);
949 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
951 osc_update_grant(cli, body);
957 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
959 client_obd_list_lock(&cli->cl_loi_list_lock);
960 oa->o_grant = cli->cl_avail_grant / 4;
961 cli->cl_avail_grant -= oa->o_grant;
962 client_obd_list_unlock(&cli->cl_loi_list_lock);
963 oa->o_flags |= OBD_FL_SHRINK_GRANT;
964 osc_update_next_shrink(cli);
967 static int osc_shrink_grant(struct client_obd *cli)
970 struct ost_body *body;
977 osc_announce_cached(cli, &body->oa, 0);
978 osc_shrink_grant_local(cli, &body->oa);
979 rc = osc_set_info_async(cli->cl_import->imp_obd->obd_self_export,
980 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
981 sizeof(*body), body, NULL);
983 __osc_update_grant(cli, body->oa.o_grant);
989 #define GRANT_SHRINK_LIMIT PTLRPC_MAX_BRW_SIZE
990 static int osc_should_shrink_grant(struct client_obd *client)
992 cfs_time_t time = cfs_time_current();
993 cfs_time_t next_shrink = client->cl_next_shrink_grant;
994 if (cfs_time_aftereq(time, next_shrink - 5 * CFS_TICK)) {
995 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
996 client->cl_avail_grant > GRANT_SHRINK_LIMIT)
999 osc_update_next_shrink(client);
1004 static int osc_grant_shrink_grant_cb(struct timeout_item *item, void *data)
1006 struct client_obd *client;
1008 list_for_each_entry(client, &item->ti_obd_list, cl_grant_shrink_list) {
1009 if (osc_should_shrink_grant(client))
1010 osc_shrink_grant(client);
1015 static int osc_add_shrink_grant(struct client_obd *client)
1019 rc = ptlrpc_add_timeout_client(GRANT_SHRINK_INTERVAL,
1021 osc_grant_shrink_grant_cb, NULL,
1022 &client->cl_grant_shrink_list);
1024 CERROR("add grant client %s error %d\n",
1025 client->cl_import->imp_obd->obd_name, rc);
1028 CDEBUG(D_CACHE, "add grant client %s \n",
1029 client->cl_import->imp_obd->obd_name);
1030 osc_update_next_shrink(client);
1034 static int osc_del_shrink_grant(struct client_obd *client)
1036 return ptlrpc_del_timeout_client(&client->cl_grant_shrink_list,
1040 static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1042 client_obd_list_lock(&cli->cl_loi_list_lock);
1043 cli->cl_avail_grant = ocd->ocd_grant;
1044 client_obd_list_unlock(&cli->cl_loi_list_lock);
1046 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1047 list_empty(&cli->cl_grant_shrink_list))
1048 osc_add_shrink_grant(cli);
1050 CDEBUG(D_CACHE, "setting cl_avail_grant: %ld cl_lost_grant: %ld \n",
1051 cli->cl_avail_grant, cli->cl_lost_grant);
1052 LASSERT(cli->cl_avail_grant >= 0);
1055 /* We assume that the reason this OSC got a short read is because it read
1056 * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1057 * via the LOV, and it _knows_ it's reading inside the file, it's just that
1058 * this stripe never got written at or beyond this stripe offset yet. */
1059 static void handle_short_read(int nob_read, obd_count page_count,
1060 struct brw_page **pga)
1065 /* skip bytes read OK */
1066 while (nob_read > 0) {
1067 LASSERT (page_count > 0);
1069 if (pga[i]->count > nob_read) {
1070 /* EOF inside this page */
1071 ptr = cfs_kmap(pga[i]->pg) +
1072 (pga[i]->off & ~CFS_PAGE_MASK);
1073 memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1074 cfs_kunmap(pga[i]->pg);
1080 nob_read -= pga[i]->count;
1085 /* zero remaining pages */
1086 while (page_count-- > 0) {
1087 ptr = cfs_kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1088 memset(ptr, 0, pga[i]->count);
1089 cfs_kunmap(pga[i]->pg);
1094 static int check_write_rcs(struct ptlrpc_request *req,
1095 int requested_nob, int niocount,
1096 obd_count page_count, struct brw_page **pga)
1100 /* return error if any niobuf was in error */
1101 remote_rcs = lustre_swab_repbuf(req, REQ_REC_OFF + 1,
1102 sizeof(*remote_rcs) * niocount, NULL);
1103 if (remote_rcs == NULL) {
1104 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1107 if (lustre_msg_swabbed(req->rq_repmsg))
1108 for (i = 0; i < niocount; i++)
1109 __swab32s(&remote_rcs[i]);
1111 for (i = 0; i < niocount; i++) {
1112 if (remote_rcs[i] < 0)
1113 return(remote_rcs[i]);
1115 if (remote_rcs[i] != 0) {
1116 CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1117 i, remote_rcs[i], req);
1122 if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1123 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1124 req->rq_bulk->bd_nob_transferred, requested_nob);
1131 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1133 if (p1->flag != p2->flag) {
1134 unsigned mask = ~(OBD_BRW_FROM_GRANT|
1135 OBD_BRW_NOCACHE|OBD_BRW_SYNC);
1137 /* warn if we try to combine flags that we don't know to be
1138 * safe to combine */
1139 if ((p1->flag & mask) != (p2->flag & mask))
1140 CERROR("is it ok to have flags 0x%x and 0x%x in the "
1141 "same brw?\n", p1->flag, p2->flag);
1145 return (p1->off + p1->count == p2->off);
1148 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1149 struct brw_page **pga, int opc,
1150 cksum_type_t cksum_type)
1155 LASSERT (pg_count > 0);
1156 cksum = init_checksum(cksum_type);
1157 while (nob > 0 && pg_count > 0) {
1158 unsigned char *ptr = cfs_kmap(pga[i]->pg);
1159 int off = pga[i]->off & ~CFS_PAGE_MASK;
1160 int count = pga[i]->count > nob ? nob : pga[i]->count;
1162 /* corrupt the data before we compute the checksum, to
1163 * simulate an OST->client data error */
1164 if (i == 0 && opc == OST_READ &&
1165 OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))
1166 memcpy(ptr + off, "bad1", min(4, nob));
1167 cksum = compute_checksum(cksum, ptr + off, count, cksum_type);
1168 cfs_kunmap(pga[i]->pg);
1169 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d checksum %x\n",
1172 nob -= pga[i]->count;
1176 /* For sending we only compute the wrong checksum instead
1177 * of corrupting the data so it is still correct on a redo */
1178 if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1184 static int osc_brw_prep_request(int cmd, struct client_obd *cli,struct obdo *oa,
1185 struct lov_stripe_md *lsm, obd_count page_count,
1186 struct brw_page **pga,
1187 struct ptlrpc_request **reqp,
1188 struct obd_capa *ocapa, int reserve)
1190 struct ptlrpc_request *req;
1191 struct ptlrpc_bulk_desc *desc;
1192 struct ost_body *body;
1193 struct obd_ioobj *ioobj;
1194 struct niobuf_remote *niobuf;
1195 int niocount, i, requested_nob, opc, rc;
1196 struct osc_brw_async_args *aa;
1197 struct req_capsule *pill;
1198 struct brw_page *pg_prev;
1201 if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1202 RETURN(-ENOMEM); /* Recoverable */
1203 if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1204 RETURN(-EINVAL); /* Fatal */
1206 if ((cmd & OBD_BRW_WRITE) != 0) {
1208 req = ptlrpc_request_alloc_pool(cli->cl_import,
1209 cli->cl_import->imp_rq_pool,
1213 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW);
1218 for (niocount = i = 1; i < page_count; i++) {
1219 if (!can_merge_pages(pga[i - 1], pga[i]))
1223 pill = &req->rq_pill;
1224 req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1225 niocount * sizeof(*niobuf));
1226 osc_set_capa_size(req, &RMF_CAPA1, ocapa);
1228 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1230 ptlrpc_request_free(req);
1233 req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1234 ptlrpc_at_set_req_timeout(req);
1236 if (opc == OST_WRITE)
1237 desc = ptlrpc_prep_bulk_imp(req, page_count,
1238 BULK_GET_SOURCE, OST_BULK_PORTAL);
1240 desc = ptlrpc_prep_bulk_imp(req, page_count,
1241 BULK_PUT_SINK, OST_BULK_PORTAL);
1244 GOTO(out, rc = -ENOMEM);
1245 /* NB request now owns desc and will free it when it gets freed */
1247 body = req_capsule_client_get(pill, &RMF_OST_BODY);
1248 ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1249 niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1250 LASSERT(body && ioobj && niobuf);
1254 obdo_to_ioobj(oa, ioobj);
1255 ioobj->ioo_bufcnt = niocount;
1256 osc_pack_capa(req, body, ocapa);
1257 LASSERT (page_count > 0);
1259 for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1260 struct brw_page *pg = pga[i];
1262 LASSERT(pg->count > 0);
1263 LASSERTF((pg->off & ~CFS_PAGE_MASK) + pg->count <= CFS_PAGE_SIZE,
1264 "i: %d pg: %p off: "LPU64", count: %u\n", i, pg,
1265 pg->off, pg->count);
1267 LASSERTF(i == 0 || pg->off > pg_prev->off,
1268 "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
1269 " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
1271 pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1272 pg_prev->pg, page_private(pg_prev->pg),
1273 pg_prev->pg->index, pg_prev->off);
1275 LASSERTF(i == 0 || pg->off > pg_prev->off,
1276 "i %d p_c %u\n", i, page_count);
1278 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1279 (pg->flag & OBD_BRW_SRVLOCK));
1281 ptlrpc_prep_bulk_page(desc, pg->pg, pg->off & ~CFS_PAGE_MASK,
1283 requested_nob += pg->count;
1285 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1287 niobuf->len += pg->count;
1289 niobuf->offset = pg->off;
1290 niobuf->len = pg->count;
1291 niobuf->flags = pg->flag;
1296 LASSERTF((void *)(niobuf - niocount) ==
1297 lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2,
1298 niocount * sizeof(*niobuf)),
1299 "want %p - real %p\n", lustre_msg_buf(req->rq_reqmsg,
1300 REQ_REC_OFF + 2, niocount * sizeof(*niobuf)),
1301 (void *)(niobuf - niocount));
1303 osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1304 if (osc_should_shrink_grant(cli))
1305 osc_shrink_grant_local(cli, &body->oa);
1307 /* size[REQ_REC_OFF] still sizeof (*body) */
1308 if (opc == OST_WRITE) {
1309 if (unlikely(cli->cl_checksum) &&
1310 !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1311 /* store cl_cksum_type in a local variable since
1312 * it can be changed via lprocfs */
1313 cksum_type_t cksum_type = cli->cl_cksum_type;
1315 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1316 oa->o_flags = body->oa.o_flags = 0;
1317 body->oa.o_flags |= cksum_type_pack(cksum_type);
1318 body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1319 body->oa.o_cksum = osc_checksum_bulk(requested_nob,
1323 CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1325 /* save this in 'oa', too, for later checking */
1326 oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1327 oa->o_flags |= cksum_type_pack(cksum_type);
1329 /* clear out the checksum flag, in case this is a
1330 * resend but cl_checksum is no longer set. b=11238 */
1331 oa->o_valid &= ~OBD_MD_FLCKSUM;
1333 oa->o_cksum = body->oa.o_cksum;
1334 /* 1 RC per niobuf */
1335 req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_SERVER,
1336 sizeof(__u32) * niocount);
1338 if (unlikely(cli->cl_checksum) &&
1339 !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1340 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1341 body->oa.o_flags = 0;
1342 body->oa.o_flags |= cksum_type_pack(cli->cl_cksum_type);
1343 body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1345 req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_SERVER, 0);
1346 /* 1 RC for the whole I/O */
1348 ptlrpc_request_set_replen(req);
1350 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1351 aa = ptlrpc_req_async_args(req);
1353 aa->aa_requested_nob = requested_nob;
1354 aa->aa_nio_count = niocount;
1355 aa->aa_page_count = page_count;
1359 CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1360 if (ocapa && reserve)
1361 aa->aa_ocapa = capa_get(ocapa);
1367 ptlrpc_req_finished(req);
1371 static int check_write_checksum(struct obdo *oa, const lnet_process_id_t *peer,
1372 __u32 client_cksum, __u32 server_cksum, int nob,
1373 obd_count page_count, struct brw_page **pga,
1374 cksum_type_t client_cksum_type)
1378 cksum_type_t cksum_type;
1380 if (server_cksum == client_cksum) {
1381 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1385 if (oa->o_valid & OBD_MD_FLFLAGS)
1386 cksum_type = cksum_type_unpack(oa->o_flags);
1388 cksum_type = OBD_CKSUM_CRC32;
1390 new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1393 if (cksum_type != client_cksum_type)
1394 msg = "the server did not use the checksum type specified in "
1395 "the original request - likely a protocol problem";
1396 else if (new_cksum == server_cksum)
1397 msg = "changed on the client after we checksummed it - "
1398 "likely false positive due to mmap IO (bug 11742)";
1399 else if (new_cksum == client_cksum)
1400 msg = "changed in transit before arrival at OST";
1402 msg = "changed in transit AND doesn't match the original - "
1403 "likely false positive due to mmap IO (bug 11742)";
1405 LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inum "
1406 LPU64"/"LPU64" object "LPU64"/"LPU64" extent "
1407 "["LPU64"-"LPU64"]\n",
1408 msg, libcfs_nid2str(peer->nid),
1409 oa->o_valid & OBD_MD_FLFID ? oa->o_fid : (__u64)0,
1410 oa->o_valid & OBD_MD_FLFID ? oa->o_generation :
1413 oa->o_valid & OBD_MD_FLGROUP ? oa->o_gr : (__u64)0,
1415 pga[page_count-1]->off + pga[page_count-1]->count - 1);
1416 CERROR("original client csum %x (type %x), server csum %x (type %x), "
1417 "client csum now %x\n", client_cksum, client_cksum_type,
1418 server_cksum, cksum_type, new_cksum);
1422 /* Note rc enters this function as number of bytes transferred */
1423 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1425 struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1426 const lnet_process_id_t *peer =
1427 &req->rq_import->imp_connection->c_peer;
1428 struct client_obd *cli = aa->aa_cli;
1429 struct ost_body *body;
1430 __u32 client_cksum = 0;
1433 if (rc < 0 && rc != -EDQUOT)
1436 LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1437 body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
1438 lustre_swab_ost_body);
1440 CDEBUG(D_INFO, "Can't unpack body\n");
1444 /* set/clear over quota flag for a uid/gid */
1445 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1446 body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA))
1447 lquota_setdq(quota_interface, cli, body->oa.o_uid,
1448 body->oa.o_gid, body->oa.o_valid,
1454 if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1455 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1457 osc_update_grant(cli, body);
1459 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1461 CERROR("Unexpected +ve rc %d\n", rc);
1464 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1466 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1469 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1470 check_write_checksum(&body->oa, peer, client_cksum,
1471 body->oa.o_cksum, aa->aa_requested_nob,
1472 aa->aa_page_count, aa->aa_ppga,
1473 cksum_type_unpack(aa->aa_oa->o_flags)))
1476 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1477 aa->aa_page_count, aa->aa_ppga);
1481 /* The rest of this function executes only for OST_READs */
1483 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1487 if (rc > aa->aa_requested_nob) {
1488 CERROR("Unexpected rc %d (%d requested)\n", rc,
1489 aa->aa_requested_nob);
1493 if (rc != req->rq_bulk->bd_nob_transferred) {
1494 CERROR ("Unexpected rc %d (%d transferred)\n",
1495 rc, req->rq_bulk->bd_nob_transferred);
1499 if (rc < aa->aa_requested_nob)
1500 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1502 if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1503 static int cksum_counter;
1504 __u32 server_cksum = body->oa.o_cksum;
1507 cksum_type_t cksum_type;
1509 if (body->oa.o_valid & OBD_MD_FLFLAGS)
1510 cksum_type = cksum_type_unpack(body->oa.o_flags);
1512 cksum_type = OBD_CKSUM_CRC32;
1513 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1514 aa->aa_ppga, OST_READ,
1517 if (peer->nid == req->rq_bulk->bd_sender) {
1521 router = libcfs_nid2str(req->rq_bulk->bd_sender);
1524 if (server_cksum == ~0 && rc > 0) {
1525 CERROR("Protocol error: server %s set the 'checksum' "
1526 "bit, but didn't send a checksum. Not fatal, "
1527 "but please notify on http://bugzilla.lustre.org/\n",
1528 libcfs_nid2str(peer->nid));
1529 } else if (server_cksum != client_cksum) {
1530 LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1531 "%s%s%s inum "LPU64"/"LPU64" object "
1532 LPU64"/"LPU64" extent "
1533 "["LPU64"-"LPU64"]\n",
1534 req->rq_import->imp_obd->obd_name,
1535 libcfs_nid2str(peer->nid),
1537 body->oa.o_valid & OBD_MD_FLFID ?
1538 body->oa.o_fid : (__u64)0,
1539 body->oa.o_valid & OBD_MD_FLFID ?
1540 body->oa.o_generation :(__u64)0,
1542 body->oa.o_valid & OBD_MD_FLGROUP ?
1543 body->oa.o_gr : (__u64)0,
1544 aa->aa_ppga[0]->off,
1545 aa->aa_ppga[aa->aa_page_count-1]->off +
1546 aa->aa_ppga[aa->aa_page_count-1]->count -
1548 CERROR("client %x, server %x, cksum_type %x\n",
1549 client_cksum, server_cksum, cksum_type);
1551 aa->aa_oa->o_cksum = client_cksum;
1555 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1558 } else if (unlikely(client_cksum)) {
1559 static int cksum_missed;
1562 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1563 CERROR("Checksum %u requested from %s but not sent\n",
1564 cksum_missed, libcfs_nid2str(peer->nid));
1570 *aa->aa_oa = body->oa;
1575 static int osc_brw_internal(int cmd, struct obd_export *exp, struct obdo *oa,
1576 struct lov_stripe_md *lsm,
1577 obd_count page_count, struct brw_page **pga,
1578 struct obd_capa *ocapa)
1580 struct ptlrpc_request *req;
1584 struct l_wait_info lwi;
1588 cfs_waitq_init(&waitq);
1591 rc = osc_brw_prep_request(cmd, &exp->exp_obd->u.cli, oa, lsm,
1592 page_count, pga, &req, ocapa, 0);
1596 rc = ptlrpc_queue_wait(req);
1598 if (rc == -ETIMEDOUT && req->rq_resend) {
1599 DEBUG_REQ(D_HA, req, "BULK TIMEOUT");
1600 ptlrpc_req_finished(req);
1604 rc = osc_brw_fini_request(req, rc);
1606 ptlrpc_req_finished(req);
1607 if (osc_recoverable_error(rc)) {
1609 if (!osc_should_resend(resends, &exp->exp_obd->u.cli)) {
1610 CERROR("too many resend retries, returning error\n");
1614 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1615 l_wait_event(waitq, 0, &lwi);
1623 int osc_brw_redo_request(struct ptlrpc_request *request,
1624 struct osc_brw_async_args *aa)
1626 struct ptlrpc_request *new_req;
1627 struct ptlrpc_request_set *set = request->rq_set;
1628 struct osc_brw_async_args *new_aa;
1629 struct osc_async_page *oap;
1633 if (!osc_should_resend(aa->aa_resends, aa->aa_cli)) {
1634 CERROR("too many resend retries, returning error\n");
1638 DEBUG_REQ(D_ERROR, request, "redo for recoverable error");
1640 rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1641 OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1642 aa->aa_cli, aa->aa_oa,
1643 NULL /* lsm unused by osc currently */,
1644 aa->aa_page_count, aa->aa_ppga,
1645 &new_req, aa->aa_ocapa, 0);
1649 client_obd_list_lock(&aa->aa_cli->cl_loi_list_lock);
1651 list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1652 if (oap->oap_request != NULL) {
1653 LASSERTF(request == oap->oap_request,
1654 "request %p != oap_request %p\n",
1655 request, oap->oap_request);
1656 if (oap->oap_interrupted) {
1657 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1658 ptlrpc_req_finished(new_req);
1663 /* New request takes over pga and oaps from old request.
1664 * Note that copying a list_head doesn't work, need to move it... */
1666 new_req->rq_interpret_reply = request->rq_interpret_reply;
1667 new_req->rq_async_args = request->rq_async_args;
1668 new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1670 new_aa = ptlrpc_req_async_args(new_req);
1672 CFS_INIT_LIST_HEAD(&new_aa->aa_oaps);
1673 list_splice(&aa->aa_oaps, &new_aa->aa_oaps);
1674 CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1676 list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1677 if (oap->oap_request) {
1678 ptlrpc_req_finished(oap->oap_request);
1679 oap->oap_request = ptlrpc_request_addref(new_req);
1683 new_aa->aa_ocapa = aa->aa_ocapa;
1684 aa->aa_ocapa = NULL;
1686 /* use ptlrpc_set_add_req is safe because interpret functions work
1687 * in check_set context. only one way exist with access to request
1688 * from different thread got -EINTR - this way protected with
1689 * cl_loi_list_lock */
1690 ptlrpc_set_add_req(set, new_req);
1692 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1694 DEBUG_REQ(D_INFO, new_req, "new request");
1699 * ugh, we want disk allocation on the target to happen in offset order. we'll
1700 * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1701 * fine for our small page arrays and doesn't require allocation. its an
1702 * insertion sort that swaps elements that are strides apart, shrinking the
1703 * stride down until its '1' and the array is sorted.
1705 static void sort_brw_pages(struct brw_page **array, int num)
1708 struct brw_page *tmp;
1712 for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1717 for (i = stride ; i < num ; i++) {
1720 while (j >= stride && array[j - stride]->off > tmp->off) {
1721 array[j] = array[j - stride];
1726 } while (stride > 1);
1729 static obd_count max_unfragmented_pages(struct brw_page **pg, obd_count pages)
1735 LASSERT (pages > 0);
1736 offset = pg[i]->off & ~CFS_PAGE_MASK;
1740 if (pages == 0) /* that's all */
1743 if (offset + pg[i]->count < CFS_PAGE_SIZE)
1744 return count; /* doesn't end on page boundary */
1747 offset = pg[i]->off & ~CFS_PAGE_MASK;
1748 if (offset != 0) /* doesn't start on page boundary */
1755 static struct brw_page **osc_build_ppga(struct brw_page *pga, obd_count count)
1757 struct brw_page **ppga;
1760 OBD_ALLOC(ppga, sizeof(*ppga) * count);
1764 for (i = 0; i < count; i++)
1769 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1771 LASSERT(ppga != NULL);
1772 OBD_FREE(ppga, sizeof(*ppga) * count);
1775 static int osc_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1776 obd_count page_count, struct brw_page *pga,
1777 struct obd_trans_info *oti)
1779 struct obdo *saved_oa = NULL;
1780 struct brw_page **ppga, **orig;
1781 struct obd_import *imp = class_exp2cliimp(exp);
1782 struct client_obd *cli;
1783 int rc, page_count_orig;
1786 LASSERT((imp != NULL) && (imp->imp_obd != NULL));
1787 cli = &imp->imp_obd->u.cli;
1789 if (cmd & OBD_BRW_CHECK) {
1790 /* The caller just wants to know if there's a chance that this
1791 * I/O can succeed */
1793 if (imp->imp_invalid)
1798 /* test_brw with a failed create can trip this, maybe others. */
1799 LASSERT(cli->cl_max_pages_per_rpc);
1803 orig = ppga = osc_build_ppga(pga, page_count);
1806 page_count_orig = page_count;
1808 sort_brw_pages(ppga, page_count);
1809 while (page_count) {
1810 obd_count pages_per_brw;
1812 if (page_count > cli->cl_max_pages_per_rpc)
1813 pages_per_brw = cli->cl_max_pages_per_rpc;
1815 pages_per_brw = page_count;
1817 pages_per_brw = max_unfragmented_pages(ppga, pages_per_brw);
1819 if (saved_oa != NULL) {
1820 /* restore previously saved oa */
1821 *oinfo->oi_oa = *saved_oa;
1822 } else if (page_count > pages_per_brw) {
1823 /* save a copy of oa (brw will clobber it) */
1824 OBDO_ALLOC(saved_oa);
1825 if (saved_oa == NULL)
1826 GOTO(out, rc = -ENOMEM);
1827 *saved_oa = *oinfo->oi_oa;
1830 rc = osc_brw_internal(cmd, exp, oinfo->oi_oa, oinfo->oi_md,
1831 pages_per_brw, ppga, oinfo->oi_capa);
1836 page_count -= pages_per_brw;
1837 ppga += pages_per_brw;
1841 osc_release_ppga(orig, page_count_orig);
1843 if (saved_oa != NULL)
1844 OBDO_FREE(saved_oa);
1849 /* The companion to osc_enter_cache(), called when @oap is no longer part of
1850 * the dirty accounting. Writeback completes or truncate happens before
1851 * writing starts. Must be called with the loi lock held. */
1852 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1855 osc_release_write_grant(cli, &oap->oap_brw_page, sent);
1859 /* This maintains the lists of pending pages to read/write for a given object
1860 * (lop). This is used by osc_check_rpcs->osc_next_loi() and loi_list_maint()
1861 * to quickly find objects that are ready to send an RPC. */
1862 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1868 if (lop->lop_num_pending == 0)
1871 /* if we have an invalid import we want to drain the queued pages
1872 * by forcing them through rpcs that immediately fail and complete
1873 * the pages. recovery relies on this to empty the queued pages
1874 * before canceling the locks and evicting down the llite pages */
1875 if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1878 /* stream rpcs in queue order as long as as there is an urgent page
1879 * queued. this is our cheap solution for good batching in the case
1880 * where writepage marks some random page in the middle of the file
1881 * as urgent because of, say, memory pressure */
1882 if (!list_empty(&lop->lop_urgent)) {
1883 CDEBUG(D_CACHE, "urgent request forcing RPC\n");
1886 /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1887 optimal = cli->cl_max_pages_per_rpc;
1888 if (cmd & OBD_BRW_WRITE) {
1889 /* trigger a write rpc stream as long as there are dirtiers
1890 * waiting for space. as they're waiting, they're not going to
1891 * create more pages to coallesce with what's waiting.. */
1892 if (!list_empty(&cli->cl_cache_waiters)) {
1893 CDEBUG(D_CACHE, "cache waiters forcing RPC\n");
1896 /* +16 to avoid triggering rpcs that would want to include pages
1897 * that are being queued but which can't be made ready until
1898 * the queuer finishes with the page. this is a wart for
1899 * llite::commit_write() */
1902 if (lop->lop_num_pending >= optimal)
1908 static int lop_makes_hprpc(struct loi_oap_pages *lop)
1910 struct osc_async_page *oap;
1913 if (list_empty(&lop->lop_urgent))
1916 oap = list_entry(lop->lop_urgent.next,
1917 struct osc_async_page, oap_urgent_item);
1919 if (oap->oap_async_flags & ASYNC_HP) {
1920 CDEBUG(D_CACHE, "hp request forcing RPC\n");
1927 static void on_list(struct list_head *item, struct list_head *list,
1930 if (list_empty(item) && should_be_on)
1931 list_add_tail(item, list);
1932 else if (!list_empty(item) && !should_be_on)
1933 list_del_init(item);
1936 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1937 * can find pages to build into rpcs quickly */
1938 void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1940 if (lop_makes_hprpc(&loi->loi_write_lop) ||
1941 lop_makes_hprpc(&loi->loi_read_lop)) {
1943 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list, 0);
1944 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 1);
1946 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 0);
1947 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list,
1948 lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)||
1949 lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
1952 on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
1953 loi->loi_write_lop.lop_num_pending);
1955 on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
1956 loi->loi_read_lop.lop_num_pending);
1959 static void lop_update_pending(struct client_obd *cli,
1960 struct loi_oap_pages *lop, int cmd, int delta)
1962 lop->lop_num_pending += delta;
1963 if (cmd & OBD_BRW_WRITE)
1964 cli->cl_pending_w_pages += delta;
1966 cli->cl_pending_r_pages += delta;
1970 * this is called when a sync waiter receives an interruption. Its job is to
1971 * get the caller woken as soon as possible. If its page hasn't been put in an
1972 * rpc yet it can dequeue immediately. Otherwise it has to mark the rpc as
1973 * desiring interruption which will forcefully complete the rpc once the rpc
1976 int osc_oap_interrupted(const struct lu_env *env, struct osc_async_page *oap)
1978 struct loi_oap_pages *lop;
1979 struct lov_oinfo *loi;
1983 LASSERT(!oap->oap_interrupted);
1984 oap->oap_interrupted = 1;
1986 /* ok, it's been put in an rpc. only one oap gets a request reference */
1987 if (oap->oap_request != NULL) {
1988 ptlrpc_mark_interrupted(oap->oap_request);
1989 ptlrpcd_wake(oap->oap_request);
1990 ptlrpc_req_finished(oap->oap_request);
1991 oap->oap_request = NULL;
1995 * page completion may be called only if ->cpo_prep() method was
1996 * executed by osc_io_submit(), that also adds page the to pending list
1998 if (!list_empty(&oap->oap_pending_item)) {
1999 list_del_init(&oap->oap_pending_item);
2000 list_del_init(&oap->oap_urgent_item);
2003 lop = (oap->oap_cmd & OBD_BRW_WRITE) ?
2004 &loi->loi_write_lop : &loi->loi_read_lop;
2005 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
2006 loi_list_maint(oap->oap_cli, oap->oap_loi);
2007 rc = oap->oap_caller_ops->ap_completion(env,
2008 oap->oap_caller_data,
2009 oap->oap_cmd, NULL, -EINTR);
2015 /* this is trying to propogate async writeback errors back up to the
2016 * application. As an async write fails we record the error code for later if
2017 * the app does an fsync. As long as errors persist we force future rpcs to be
2018 * sync so that the app can get a sync error and break the cycle of queueing
2019 * pages for which writeback will fail. */
2020 static void osc_process_ar(struct osc_async_rc *ar, __u64 xid,
2027 ar->ar_force_sync = 1;
2028 ar->ar_min_xid = ptlrpc_sample_next_xid();
2033 if (ar->ar_force_sync && (xid >= ar->ar_min_xid))
2034 ar->ar_force_sync = 0;
2037 void osc_oap_to_pending(struct osc_async_page *oap)
2039 struct loi_oap_pages *lop;
2041 if (oap->oap_cmd & OBD_BRW_WRITE)
2042 lop = &oap->oap_loi->loi_write_lop;
2044 lop = &oap->oap_loi->loi_read_lop;
2046 if (oap->oap_async_flags & ASYNC_HP)
2047 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2048 else if (oap->oap_async_flags & ASYNC_URGENT)
2049 list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2050 list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2051 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, 1);
2054 /* this must be called holding the loi list lock to give coverage to exit_cache,
2055 * async_flag maintenance, and oap_request */
2056 static void osc_ap_completion(const struct lu_env *env,
2057 struct client_obd *cli, struct obdo *oa,
2058 struct osc_async_page *oap, int sent, int rc)
2063 if (oap->oap_request != NULL) {
2064 xid = ptlrpc_req_xid(oap->oap_request);
2065 ptlrpc_req_finished(oap->oap_request);
2066 oap->oap_request = NULL;
2069 oap->oap_async_flags = 0;
2070 oap->oap_interrupted = 0;
2072 if (oap->oap_cmd & OBD_BRW_WRITE) {
2073 osc_process_ar(&cli->cl_ar, xid, rc);
2074 osc_process_ar(&oap->oap_loi->loi_ar, xid, rc);
2077 if (rc == 0 && oa != NULL) {
2078 if (oa->o_valid & OBD_MD_FLBLOCKS)
2079 oap->oap_loi->loi_lvb.lvb_blocks = oa->o_blocks;
2080 if (oa->o_valid & OBD_MD_FLMTIME)
2081 oap->oap_loi->loi_lvb.lvb_mtime = oa->o_mtime;
2082 if (oa->o_valid & OBD_MD_FLATIME)
2083 oap->oap_loi->loi_lvb.lvb_atime = oa->o_atime;
2084 if (oa->o_valid & OBD_MD_FLCTIME)
2085 oap->oap_loi->loi_lvb.lvb_ctime = oa->o_ctime;
2088 rc = oap->oap_caller_ops->ap_completion(env, oap->oap_caller_data,
2089 oap->oap_cmd, oa, rc);
2091 /* ll_ap_completion (from llite) drops PG_locked. so, a new
2092 * I/O on the page could start, but OSC calls it under lock
2093 * and thus we can add oap back to pending safely */
2095 /* upper layer wants to leave the page on pending queue */
2096 osc_oap_to_pending(oap);
2098 osc_exit_cache(cli, oap, sent);
2102 static int brw_interpret(const struct lu_env *env,
2103 struct ptlrpc_request *req, void *data, int rc)
2105 struct osc_brw_async_args *aa = data;
2106 struct client_obd *cli;
2110 rc = osc_brw_fini_request(req, rc);
2111 CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2112 if (osc_recoverable_error(rc)) {
2113 rc = osc_brw_redo_request(req, aa);
2119 capa_put(aa->aa_ocapa);
2120 aa->aa_ocapa = NULL;
2125 client_obd_list_lock(&cli->cl_loi_list_lock);
2127 /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2128 * is called so we know whether to go to sync BRWs or wait for more
2129 * RPCs to complete */
2130 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2131 cli->cl_w_in_flight--;
2133 cli->cl_r_in_flight--;
2135 async = list_empty(&aa->aa_oaps);
2136 if (!async) { /* from osc_send_oap_rpc() */
2137 struct osc_async_page *oap, *tmp;
2138 /* the caller may re-use the oap after the completion call so
2139 * we need to clean it up a little */
2140 list_for_each_entry_safe(oap, tmp, &aa->aa_oaps, oap_rpc_item) {
2141 list_del_init(&oap->oap_rpc_item);
2142 osc_ap_completion(env, cli, aa->aa_oa, oap, 1, rc);
2144 OBDO_FREE(aa->aa_oa);
2145 } else { /* from async_internal() */
2147 for (i = 0; i < aa->aa_page_count; i++)
2148 osc_release_write_grant(aa->aa_cli, aa->aa_ppga[i], 1);
2150 osc_wake_cache_waiters(cli);
2151 osc_check_rpcs(env, cli);
2152 client_obd_list_unlock(&cli->cl_loi_list_lock);
2154 cl_req_completion(env, aa->aa_clerq, rc);
2155 osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2159 static struct ptlrpc_request *osc_build_req(const struct lu_env *env,
2160 struct client_obd *cli,
2161 struct list_head *rpc_list,
2162 int page_count, int cmd)
2164 struct ptlrpc_request *req;
2165 struct brw_page **pga = NULL;
2166 struct osc_brw_async_args *aa;
2167 struct obdo *oa = NULL;
2168 const struct obd_async_page_ops *ops = NULL;
2169 void *caller_data = NULL;
2170 struct osc_async_page *oap;
2171 struct osc_async_page *tmp;
2172 struct ost_body *body;
2173 struct cl_req *clerq = NULL;
2174 enum cl_req_type crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2175 struct ldlm_lock *lock = NULL;
2176 struct cl_req_attr crattr;
2180 LASSERT(!list_empty(rpc_list));
2182 memset(&crattr, 0, sizeof crattr);
2183 OBD_ALLOC(pga, sizeof(*pga) * page_count);
2185 GOTO(out, req = ERR_PTR(-ENOMEM));
2189 GOTO(out, req = ERR_PTR(-ENOMEM));
2192 list_for_each_entry(oap, rpc_list, oap_rpc_item) {
2193 struct cl_page *page = osc_oap2cl_page(oap);
2195 ops = oap->oap_caller_ops;
2196 caller_data = oap->oap_caller_data;
2198 clerq = cl_req_alloc(env, page, crt,
2199 1 /* only 1-object rpcs for
2202 GOTO(out, req = (void *)clerq);
2203 lock = oap->oap_ldlm_lock;
2205 pga[i] = &oap->oap_brw_page;
2206 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2207 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
2208 pga[i]->pg, cfs_page_index(oap->oap_page), oap, pga[i]->flag);
2210 cl_req_page_add(env, clerq, page);
2213 /* always get the data for the obdo for the rpc */
2214 LASSERT(ops != NULL);
2216 crattr.cra_capa = NULL;
2217 cl_req_attr_set(env, clerq, &crattr, ~0ULL);
2219 oa->o_handle = lock->l_remote_handle;
2220 oa->o_valid |= OBD_MD_FLHANDLE;
2223 rc = cl_req_prep(env, clerq);
2225 CERROR("cl_req_prep failed: %d\n", rc);
2226 GOTO(out, req = ERR_PTR(rc));
2229 sort_brw_pages(pga, page_count);
2230 rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
2231 pga, &req, crattr.cra_capa, 1);
2233 CERROR("prep_req failed: %d\n", rc);
2234 GOTO(out, req = ERR_PTR(rc));
2237 /* Need to update the timestamps after the request is built in case
2238 * we race with setattr (locally or in queue at OST). If OST gets
2239 * later setattr before earlier BRW (as determined by the request xid),
2240 * the OST will not use BRW timestamps. Sadly, there is no obvious
2241 * way to do this in a single call. bug 10150 */
2242 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2243 cl_req_attr_set(env, clerq, &crattr,
2244 OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2246 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2247 aa = ptlrpc_req_async_args(req);
2248 CFS_INIT_LIST_HEAD(&aa->aa_oaps);
2249 list_splice(rpc_list, &aa->aa_oaps);
2250 CFS_INIT_LIST_HEAD(rpc_list);
2251 aa->aa_clerq = clerq;
2253 capa_put(crattr.cra_capa);
2258 OBD_FREE(pga, sizeof(*pga) * page_count);
2259 /* this should happen rarely and is pretty bad, it makes the
2260 * pending list not follow the dirty order */
2261 client_obd_list_lock(&cli->cl_loi_list_lock);
2262 list_for_each_entry_safe(oap, tmp, rpc_list, oap_rpc_item) {
2263 list_del_init(&oap->oap_rpc_item);
2265 /* queued sync pages can be torn down while the pages
2266 * were between the pending list and the rpc */
2267 if (oap->oap_interrupted) {
2268 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
2269 osc_ap_completion(env, cli, NULL, oap, 0,
2273 osc_ap_completion(env, cli, NULL, oap, 0, PTR_ERR(req));
2275 if (clerq && !IS_ERR(clerq))
2276 cl_req_completion(env, clerq, PTR_ERR(req));
2282 * prepare pages for ASYNC io and put pages in send queue.
2286 * \param cmd - OBD_BRW_* macroses
2287 * \param lop - pending pages
2289 * \return zero if pages successfully add to send queue.
2290 * \return not zere if error occurring.
2293 osc_send_oap_rpc(const struct lu_env *env, struct client_obd *cli,
2294 struct lov_oinfo *loi,
2295 int cmd, struct loi_oap_pages *lop)
2297 struct ptlrpc_request *req;
2298 obd_count page_count = 0;
2299 struct osc_async_page *oap = NULL, *tmp;
2300 struct osc_brw_async_args *aa;
2301 const struct obd_async_page_ops *ops;
2302 CFS_LIST_HEAD(rpc_list);
2303 unsigned int ending_offset;
2304 unsigned starting_offset = 0;
2306 struct cl_object *clob = NULL;
2309 /* If there are HP OAPs we need to handle at least 1 of them,
2310 * move it the beginning of the pending list for that. */
2311 if (!list_empty(&lop->lop_urgent)) {
2312 oap = list_entry(lop->lop_urgent.next,
2313 struct osc_async_page, oap_urgent_item);
2314 if (oap->oap_async_flags & ASYNC_HP)
2315 list_move(&oap->oap_pending_item, &lop->lop_pending);
2318 /* first we find the pages we're allowed to work with */
2319 list_for_each_entry_safe(oap, tmp, &lop->lop_pending,
2321 ops = oap->oap_caller_ops;
2323 LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
2324 "magic 0x%x\n", oap, oap->oap_magic);
2327 /* pin object in memory, so that completion call-backs
2328 * can be safely called under client_obd_list lock. */
2329 clob = osc_oap2cl_page(oap)->cp_obj;
2330 cl_object_get(clob);
2333 if (page_count != 0 &&
2334 srvlock != !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK)) {
2335 CDEBUG(D_PAGE, "SRVLOCK flag mismatch,"
2336 " oap %p, page %p, srvlock %u\n",
2337 oap, oap->oap_brw_page.pg, (unsigned)!srvlock);
2340 /* in llite being 'ready' equates to the page being locked
2341 * until completion unlocks it. commit_write submits a page
2342 * as not ready because its unlock will happen unconditionally
2343 * as the call returns. if we race with commit_write giving
2344 * us that page we dont' want to create a hole in the page
2345 * stream, so we stop and leave the rpc to be fired by
2346 * another dirtier or kupdated interval (the not ready page
2347 * will still be on the dirty list). we could call in
2348 * at the end of ll_file_write to process the queue again. */
2349 if (!(oap->oap_async_flags & ASYNC_READY)) {
2350 int rc = ops->ap_make_ready(env, oap->oap_caller_data,
2353 CDEBUG(D_INODE, "oap %p page %p returned %d "
2354 "instead of ready\n", oap,
2358 /* llite is telling us that the page is still
2359 * in commit_write and that we should try
2360 * and put it in an rpc again later. we
2361 * break out of the loop so we don't create
2362 * a hole in the sequence of pages in the rpc
2367 /* the io isn't needed.. tell the checks
2368 * below to complete the rpc with EINTR */
2369 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
2370 oap->oap_count = -EINTR;
2373 oap->oap_async_flags |= ASYNC_READY;
2376 LASSERTF(0, "oap %p page %p returned %d "
2377 "from make_ready\n", oap,
2385 * Page submitted for IO has to be locked. Either by
2386 * ->ap_make_ready() or by higher layers.
2388 #if defined(__KERNEL__) && defined(__linux__)
2390 struct cl_page *page;
2392 page = osc_oap2cl_page(oap);
2394 if (page->cp_type == CPT_CACHEABLE &&
2395 !(PageLocked(oap->oap_page) &&
2396 (CheckWriteback(oap->oap_page, cmd)))) {
2397 CDEBUG(D_PAGE, "page %p lost wb %lx/%x\n",
2399 (long)oap->oap_page->flags,
2400 oap->oap_async_flags);
2405 /* If there is a gap at the start of this page, it can't merge
2406 * with any previous page, so we'll hand the network a
2407 * "fragmented" page array that it can't transfer in 1 RDMA */
2408 if (page_count != 0 && oap->oap_page_off != 0)
2411 /* take the page out of our book-keeping */
2412 list_del_init(&oap->oap_pending_item);
2413 lop_update_pending(cli, lop, cmd, -1);
2414 list_del_init(&oap->oap_urgent_item);
2416 if (page_count == 0)
2417 starting_offset = (oap->oap_obj_off+oap->oap_page_off) &
2418 (PTLRPC_MAX_BRW_SIZE - 1);
2420 /* ask the caller for the size of the io as the rpc leaves. */
2421 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE)) {
2423 ops->ap_refresh_count(env, oap->oap_caller_data,
2425 LASSERT(oap->oap_page_off + oap->oap_count <= CFS_PAGE_SIZE);
2427 if (oap->oap_count <= 0) {
2428 CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
2430 osc_ap_completion(env, cli, NULL,
2431 oap, 0, oap->oap_count);
2435 /* now put the page back in our accounting */
2436 list_add_tail(&oap->oap_rpc_item, &rpc_list);
2437 if (page_count == 0)
2438 srvlock = !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK);
2439 if (++page_count >= cli->cl_max_pages_per_rpc)
2442 /* End on a PTLRPC_MAX_BRW_SIZE boundary. We want full-sized
2443 * RPCs aligned on PTLRPC_MAX_BRW_SIZE boundaries to help reads
2444 * have the same alignment as the initial writes that allocated
2445 * extents on the server. */
2446 ending_offset = (oap->oap_obj_off + oap->oap_page_off +
2447 oap->oap_count) & (PTLRPC_MAX_BRW_SIZE - 1);
2448 if (ending_offset == 0)
2451 /* If there is a gap at the end of this page, it can't merge
2452 * with any subsequent pages, so we'll hand the network a
2453 * "fragmented" page array that it can't transfer in 1 RDMA */
2454 if (oap->oap_page_off + oap->oap_count < CFS_PAGE_SIZE)
2458 osc_wake_cache_waiters(cli);
2460 loi_list_maint(cli, loi);
2462 client_obd_list_unlock(&cli->cl_loi_list_lock);
2465 cl_object_put(env, clob);
2467 if (page_count == 0) {
2468 client_obd_list_lock(&cli->cl_loi_list_lock);
2472 req = osc_build_req(env, cli, &rpc_list, page_count, cmd);
2474 LASSERT(list_empty(&rpc_list));
2475 loi_list_maint(cli, loi);
2476 RETURN(PTR_ERR(req));
2479 aa = ptlrpc_req_async_args(req);
2481 if (cmd == OBD_BRW_READ) {
2482 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2483 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2484 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2485 (starting_offset >> CFS_PAGE_SHIFT) + 1);
2487 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2488 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
2489 cli->cl_w_in_flight);
2490 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2491 (starting_offset >> CFS_PAGE_SHIFT) + 1);
2493 ptlrpc_lprocfs_brw(req, aa->aa_requested_nob);
2495 client_obd_list_lock(&cli->cl_loi_list_lock);
2497 if (cmd == OBD_BRW_READ)
2498 cli->cl_r_in_flight++;
2500 cli->cl_w_in_flight++;
2502 /* queued sync pages can be torn down while the pages
2503 * were between the pending list and the rpc */
2505 list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2506 /* only one oap gets a request reference */
2509 if (oap->oap_interrupted && !req->rq_intr) {
2510 CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2512 ptlrpc_mark_interrupted(req);
2516 tmp->oap_request = ptlrpc_request_addref(req);
2518 DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2519 page_count, aa, cli->cl_r_in_flight, cli->cl_w_in_flight);
2521 req->rq_interpret_reply = brw_interpret;
2522 ptlrpcd_add_req(req, PSCOPE_BRW);
2526 #define LOI_DEBUG(LOI, STR, args...) \
2527 CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR, \
2528 !list_empty(&(LOI)->loi_ready_item) || \
2529 !list_empty(&(LOI)->loi_hp_ready_item), \
2530 (LOI)->loi_write_lop.lop_num_pending, \
2531 !list_empty(&(LOI)->loi_write_lop.lop_urgent), \
2532 (LOI)->loi_read_lop.lop_num_pending, \
2533 !list_empty(&(LOI)->loi_read_lop.lop_urgent), \
2536 /* This is called by osc_check_rpcs() to find which objects have pages that
2537 * we could be sending. These lists are maintained by lop_makes_rpc(). */
2538 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
2542 /* First return objects that have blocked locks so that they
2543 * will be flushed quickly and other clients can get the lock,
2544 * then objects which have pages ready to be stuffed into RPCs */
2545 if (!list_empty(&cli->cl_loi_hp_ready_list))
2546 RETURN(list_entry(cli->cl_loi_hp_ready_list.next,
2547 struct lov_oinfo, loi_hp_ready_item));
2548 if (!list_empty(&cli->cl_loi_ready_list))
2549 RETURN(list_entry(cli->cl_loi_ready_list.next,
2550 struct lov_oinfo, loi_ready_item));
2552 /* then if we have cache waiters, return all objects with queued
2553 * writes. This is especially important when many small files
2554 * have filled up the cache and not been fired into rpcs because
2555 * they don't pass the nr_pending/object threshhold */
2556 if (!list_empty(&cli->cl_cache_waiters) &&
2557 !list_empty(&cli->cl_loi_write_list))
2558 RETURN(list_entry(cli->cl_loi_write_list.next,
2559 struct lov_oinfo, loi_write_item));
2561 /* then return all queued objects when we have an invalid import
2562 * so that they get flushed */
2563 if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
2564 if (!list_empty(&cli->cl_loi_write_list))
2565 RETURN(list_entry(cli->cl_loi_write_list.next,
2566 struct lov_oinfo, loi_write_item));
2567 if (!list_empty(&cli->cl_loi_read_list))
2568 RETURN(list_entry(cli->cl_loi_read_list.next,
2569 struct lov_oinfo, loi_read_item));
2574 static int osc_max_rpc_in_flight(struct client_obd *cli, struct lov_oinfo *loi)
2576 struct osc_async_page *oap;
2579 if (!list_empty(&loi->loi_write_lop.lop_urgent)) {
2580 oap = list_entry(loi->loi_write_lop.lop_urgent.next,
2581 struct osc_async_page, oap_urgent_item);
2582 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2585 if (!hprpc && !list_empty(&loi->loi_read_lop.lop_urgent)) {
2586 oap = list_entry(loi->loi_write_lop.lop_urgent.next,
2587 struct osc_async_page, oap_urgent_item);
2588 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2591 return rpcs_in_flight(cli) >= cli->cl_max_rpcs_in_flight + hprpc;
2594 /* called with the loi list lock held */
2595 void osc_check_rpcs(const struct lu_env *env, struct client_obd *cli)
2597 struct lov_oinfo *loi;
2598 int rc = 0, race_counter = 0;
2601 while ((loi = osc_next_loi(cli)) != NULL) {
2602 LOI_DEBUG(loi, "%lu in flight\n", rpcs_in_flight(cli));
2604 if (osc_max_rpc_in_flight(cli, loi))
2607 /* attempt some read/write balancing by alternating between
2608 * reads and writes in an object. The makes_rpc checks here
2609 * would be redundant if we were getting read/write work items
2610 * instead of objects. we don't want send_oap_rpc to drain a
2611 * partial read pending queue when we're given this object to
2612 * do io on writes while there are cache waiters */
2613 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
2614 rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_WRITE,
2615 &loi->loi_write_lop);
2623 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
2624 rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_READ,
2625 &loi->loi_read_lop);
2634 /* attempt some inter-object balancing by issueing rpcs
2635 * for each object in turn */
2636 if (!list_empty(&loi->loi_hp_ready_item))
2637 list_del_init(&loi->loi_hp_ready_item);
2638 if (!list_empty(&loi->loi_ready_item))
2639 list_del_init(&loi->loi_ready_item);
2640 if (!list_empty(&loi->loi_write_item))
2641 list_del_init(&loi->loi_write_item);
2642 if (!list_empty(&loi->loi_read_item))
2643 list_del_init(&loi->loi_read_item);
2645 loi_list_maint(cli, loi);
2647 /* send_oap_rpc fails with 0 when make_ready tells it to
2648 * back off. llite's make_ready does this when it tries
2649 * to lock a page queued for write that is already locked.
2650 * we want to try sending rpcs from many objects, but we
2651 * don't want to spin failing with 0. */
2652 if (race_counter == 10)
2658 /* we're trying to queue a page in the osc so we're subject to the
2659 * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
2660 * If the osc's queued pages are already at that limit, then we want to sleep
2661 * until there is space in the osc's queue for us. We also may be waiting for
2662 * write credits from the OST if there are RPCs in flight that may return some
2663 * before we fall back to sync writes.
2665 * We need this know our allocation was granted in the presence of signals */
2666 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
2670 client_obd_list_lock(&cli->cl_loi_list_lock);
2671 rc = list_empty(&ocw->ocw_entry) || rpcs_in_flight(cli) == 0;
2672 client_obd_list_unlock(&cli->cl_loi_list_lock);
2677 * Non-blocking version of osc_enter_cache() that consumes grant only when it
2680 int osc_enter_cache_try(const struct lu_env *env,
2681 struct client_obd *cli, struct lov_oinfo *loi,
2682 struct osc_async_page *oap, int transient)
2686 has_grant = cli->cl_avail_grant >= CFS_PAGE_SIZE;
2688 osc_consume_write_grant(cli, &oap->oap_brw_page);
2690 cli->cl_dirty_transit += CFS_PAGE_SIZE;
2691 atomic_inc(&obd_dirty_transit_pages);
2692 oap->oap_brw_flags |= OBD_BRW_NOCACHE;
2698 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
2699 * grant or cache space. */
2700 static int osc_enter_cache(const struct lu_env *env,
2701 struct client_obd *cli, struct lov_oinfo *loi,
2702 struct osc_async_page *oap)
2704 struct osc_cache_waiter ocw;
2705 struct l_wait_info lwi = { 0 };
2709 CDEBUG(D_CACHE, "dirty: %ld/%d dirty_max: %ld/%d dropped: %lu "
2710 "grant: %lu\n", cli->cl_dirty, atomic_read(&obd_dirty_pages),
2711 cli->cl_dirty_max, obd_max_dirty_pages,
2712 cli->cl_lost_grant, cli->cl_avail_grant);
2714 /* force the caller to try sync io. this can jump the list
2715 * of queued writes and create a discontiguous rpc stream */
2716 if (cli->cl_dirty_max < CFS_PAGE_SIZE || cli->cl_ar.ar_force_sync ||
2717 loi->loi_ar.ar_force_sync)
2720 /* Hopefully normal case - cache space and write credits available */
2721 if (cli->cl_dirty + CFS_PAGE_SIZE <= cli->cl_dirty_max &&
2722 atomic_read(&obd_dirty_pages) + 1 <= obd_max_dirty_pages &&
2723 osc_enter_cache_try(env, cli, loi, oap, 0))
2726 /* Make sure that there are write rpcs in flight to wait for. This
2727 * is a little silly as this object may not have any pending but
2728 * other objects sure might. */
2729 if (cli->cl_w_in_flight) {
2730 list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
2731 cfs_waitq_init(&ocw.ocw_waitq);
2735 loi_list_maint(cli, loi);
2736 osc_check_rpcs(env, cli);
2737 client_obd_list_unlock(&cli->cl_loi_list_lock);
2739 CDEBUG(D_CACHE, "sleeping for cache space\n");
2740 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
2742 client_obd_list_lock(&cli->cl_loi_list_lock);
2743 if (!list_empty(&ocw.ocw_entry)) {
2744 list_del(&ocw.ocw_entry);
2754 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
2755 struct lov_oinfo *loi, cfs_page_t *page,
2756 obd_off offset, const struct obd_async_page_ops *ops,
2757 void *data, void **res, int nocache,
2758 struct lustre_handle *lockh)
2760 struct osc_async_page *oap;
2765 return size_round(sizeof(*oap));
2768 oap->oap_magic = OAP_MAGIC;
2769 oap->oap_cli = &exp->exp_obd->u.cli;
2772 oap->oap_caller_ops = ops;
2773 oap->oap_caller_data = data;
2775 oap->oap_page = page;
2776 oap->oap_obj_off = offset;
2777 if (!client_is_remote(exp) &&
2778 cfs_capable(CFS_CAP_SYS_RESOURCE))
2779 oap->oap_brw_flags = OBD_BRW_NOQUOTA;
2781 LASSERT(!(offset & ~CFS_PAGE_MASK));
2783 CFS_INIT_LIST_HEAD(&oap->oap_pending_item);
2784 CFS_INIT_LIST_HEAD(&oap->oap_urgent_item);
2785 CFS_INIT_LIST_HEAD(&oap->oap_rpc_item);
2786 CFS_INIT_LIST_HEAD(&oap->oap_page_list);
2788 spin_lock_init(&oap->oap_lock);
2789 CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
2793 struct osc_async_page *oap_from_cookie(void *cookie)
2795 struct osc_async_page *oap = cookie;
2796 if (oap->oap_magic != OAP_MAGIC)
2797 return ERR_PTR(-EINVAL);
2801 int osc_queue_async_io(const struct lu_env *env,
2802 struct obd_export *exp, struct lov_stripe_md *lsm,
2803 struct lov_oinfo *loi, void *cookie,
2804 int cmd, obd_off off, int count,
2805 obd_flag brw_flags, enum async_flags async_flags)
2807 struct client_obd *cli = &exp->exp_obd->u.cli;
2808 struct osc_async_page *oap;
2812 oap = oap_from_cookie(cookie);
2814 RETURN(PTR_ERR(oap));
2816 if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2819 if (!list_empty(&oap->oap_pending_item) ||
2820 !list_empty(&oap->oap_urgent_item) ||
2821 !list_empty(&oap->oap_rpc_item))
2824 /* check if the file's owner/group is over quota */
2825 if ((cmd & OBD_BRW_WRITE) && !(cmd & OBD_BRW_NOQUOTA)) {
2826 struct cl_object *obj;
2827 struct cl_attr attr; /* XXX put attr into thread info */
2829 obj = cl_object_top(osc_oap2cl_page(oap)->cp_obj);
2831 cl_object_attr_lock(obj);
2832 rc = cl_object_attr_get(env, obj, &attr);
2833 cl_object_attr_unlock(obj);
2835 if (rc == 0 && lquota_chkdq(quota_interface, cli, attr.cat_uid,
2836 attr.cat_gid) == NO_QUOTA)
2843 loi = lsm->lsm_oinfo[0];
2845 client_obd_list_lock(&cli->cl_loi_list_lock);
2847 LASSERT(off + count <= CFS_PAGE_SIZE);
2849 oap->oap_page_off = off;
2850 oap->oap_count = count;
2851 oap->oap_brw_flags = brw_flags;
2852 oap->oap_async_flags = async_flags;
2854 if (cmd & OBD_BRW_WRITE) {
2855 rc = osc_enter_cache(env, cli, loi, oap);
2857 client_obd_list_unlock(&cli->cl_loi_list_lock);
2862 osc_oap_to_pending(oap);
2863 loi_list_maint(cli, loi);
2865 LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
2868 osc_check_rpcs(env, cli);
2869 client_obd_list_unlock(&cli->cl_loi_list_lock);
2874 /* aka (~was & now & flag), but this is more clear :) */
2875 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
2877 int osc_set_async_flags_base(struct client_obd *cli,
2878 struct lov_oinfo *loi, struct osc_async_page *oap,
2879 obd_flag async_flags)
2881 struct loi_oap_pages *lop;
2884 if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2887 if (oap->oap_cmd & OBD_BRW_WRITE) {
2888 lop = &loi->loi_write_lop;
2890 lop = &loi->loi_read_lop;
2893 if (list_empty(&oap->oap_pending_item))
2896 if ((oap->oap_async_flags & async_flags) == async_flags)
2899 if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
2900 oap->oap_async_flags |= ASYNC_READY;
2902 if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT) &&
2903 list_empty(&oap->oap_rpc_item)) {
2904 if (oap->oap_async_flags & ASYNC_HP)
2905 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2907 list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2908 oap->oap_async_flags |= ASYNC_URGENT;
2909 loi_list_maint(cli, loi);
2912 LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
2913 oap->oap_async_flags);
2917 int osc_teardown_async_page(struct obd_export *exp,
2918 struct lov_stripe_md *lsm,
2919 struct lov_oinfo *loi, void *cookie)
2921 struct client_obd *cli = &exp->exp_obd->u.cli;
2922 struct loi_oap_pages *lop;
2923 struct osc_async_page *oap;
2927 oap = oap_from_cookie(cookie);
2929 RETURN(PTR_ERR(oap));
2932 loi = lsm->lsm_oinfo[0];
2934 if (oap->oap_cmd & OBD_BRW_WRITE) {
2935 lop = &loi->loi_write_lop;
2937 lop = &loi->loi_read_lop;
2940 client_obd_list_lock(&cli->cl_loi_list_lock);
2942 if (!list_empty(&oap->oap_rpc_item))
2943 GOTO(out, rc = -EBUSY);
2945 osc_exit_cache(cli, oap, 0);
2946 osc_wake_cache_waiters(cli);
2948 if (!list_empty(&oap->oap_urgent_item)) {
2949 list_del_init(&oap->oap_urgent_item);
2950 oap->oap_async_flags &= ~(ASYNC_URGENT | ASYNC_HP);
2952 if (!list_empty(&oap->oap_pending_item)) {
2953 list_del_init(&oap->oap_pending_item);
2954 lop_update_pending(cli, lop, oap->oap_cmd, -1);
2956 loi_list_maint(cli, loi);
2957 LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
2959 client_obd_list_unlock(&cli->cl_loi_list_lock);
2963 static void osc_set_lock_data_with_check(struct ldlm_lock *lock,
2964 struct ldlm_enqueue_info *einfo,
2967 void *data = einfo->ei_cbdata;
2969 LASSERT(lock != NULL);
2970 LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
2971 LASSERT(lock->l_resource->lr_type == einfo->ei_type);
2972 LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
2973 LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
2975 lock_res_and_lock(lock);
2976 spin_lock(&osc_ast_guard);
2977 LASSERT(lock->l_ast_data == NULL || lock->l_ast_data == data);
2978 lock->l_ast_data = data;
2979 spin_unlock(&osc_ast_guard);
2980 unlock_res_and_lock(lock);
2983 static void osc_set_data_with_check(struct lustre_handle *lockh,
2984 struct ldlm_enqueue_info *einfo,
2987 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2990 osc_set_lock_data_with_check(lock, einfo, flags);
2991 LDLM_LOCK_PUT(lock);
2993 CERROR("lockh %p, data %p - client evicted?\n",
2994 lockh, einfo->ei_cbdata);
2997 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2998 ldlm_iterator_t replace, void *data)
3000 struct ldlm_res_id res_id;
3001 struct obd_device *obd = class_exp2obd(exp);
3003 osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_gr, &res_id);
3004 ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3008 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
3009 obd_enqueue_update_f upcall, void *cookie,
3012 int intent = *flags & LDLM_FL_HAS_INTENT;
3016 /* The request was created before ldlm_cli_enqueue call. */
3017 if (rc == ELDLM_LOCK_ABORTED) {
3018 struct ldlm_reply *rep;
3019 rep = req_capsule_server_get(&req->rq_pill,
3022 LASSERT(rep != NULL);
3023 if (rep->lock_policy_res1)
3024 rc = rep->lock_policy_res1;
3028 if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3029 *flags |= LDLM_FL_LVB_READY;
3030 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3031 lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
3034 /* Call the update callback. */
3035 rc = (*upcall)(cookie, rc);
3039 static int osc_enqueue_interpret(const struct lu_env *env,
3040 struct ptlrpc_request *req,
3041 struct osc_enqueue_args *aa, int rc)
3043 struct ldlm_lock *lock;
3044 struct lustre_handle handle;
3047 /* Make a local copy of a lock handle and a mode, because aa->oa_*
3048 * might be freed anytime after lock upcall has been called. */
3049 lustre_handle_copy(&handle, aa->oa_lockh);
3050 mode = aa->oa_ei->ei_mode;
3052 /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3054 lock = ldlm_handle2lock(&handle);
3056 /* Take an additional reference so that a blocking AST that
3057 * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
3058 * to arrive after an upcall has been executed by
3059 * osc_enqueue_fini(). */
3060 ldlm_lock_addref(&handle, mode);
3062 /* Complete obtaining the lock procedure. */
3063 rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3064 mode, aa->oa_flags, aa->oa_lvb,
3065 sizeof(*aa->oa_lvb), lustre_swab_ost_lvb,
3067 /* Complete osc stuff. */
3068 rc = osc_enqueue_fini(req, aa->oa_lvb,
3069 aa->oa_upcall, aa->oa_cookie, aa->oa_flags, rc);
3071 OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
3073 /* Release the lock for async request. */
3074 if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
3076 * Releases a reference taken by ldlm_cli_enqueue(), if it is
3077 * not already released by
3078 * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
3080 ldlm_lock_decref(&handle, mode);
3082 LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3083 aa->oa_lockh, req, aa);
3084 ldlm_lock_decref(&handle, mode);
3085 LDLM_LOCK_PUT(lock);
3089 void osc_update_enqueue(struct lustre_handle *lov_lockhp,
3090 struct lov_oinfo *loi, int flags,
3091 struct ost_lvb *lvb, __u32 mode, int rc)
3093 if (rc == ELDLM_OK) {
3094 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
3097 LASSERT(lock != NULL);
3098 loi->loi_lvb = *lvb;
3099 tmp = loi->loi_lvb.lvb_size;
3100 /* Extend KMS up to the end of this lock and no further
3101 * A lock on [x,y] means a KMS of up to y + 1 bytes! */
3102 if (tmp > lock->l_policy_data.l_extent.end)
3103 tmp = lock->l_policy_data.l_extent.end + 1;
3104 if (tmp >= loi->loi_kms) {
3105 LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
3106 ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
3107 loi_kms_set(loi, tmp);
3109 LDLM_DEBUG(lock, "lock acquired, setting rss="
3110 LPU64"; leaving kms="LPU64", end="LPU64,
3111 loi->loi_lvb.lvb_size, loi->loi_kms,
3112 lock->l_policy_data.l_extent.end);
3114 ldlm_lock_allow_match(lock);
3115 LDLM_LOCK_PUT(lock);
3116 } else if (rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT)) {
3117 loi->loi_lvb = *lvb;
3118 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
3119 " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
3123 EXPORT_SYMBOL(osc_update_enqueue);
3125 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
3127 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3128 * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3129 * other synchronous requests, however keeping some locks and trying to obtain
3130 * others may take a considerable amount of time in a case of ost failure; and
3131 * when other sync requests do not get released lock from a client, the client
3132 * is excluded from the cluster -- such scenarious make the life difficult, so
3133 * release locks just after they are obtained. */
3134 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3135 int *flags, ldlm_policy_data_t *policy,
3136 struct ost_lvb *lvb, int kms_valid,
3137 obd_enqueue_update_f upcall, void *cookie,
3138 struct ldlm_enqueue_info *einfo,
3139 struct lustre_handle *lockh,
3140 struct ptlrpc_request_set *rqset, int async)
3142 struct obd_device *obd = exp->exp_obd;
3143 struct ptlrpc_request *req = NULL;
3144 int intent = *flags & LDLM_FL_HAS_INTENT;
3149 /* Filesystem lock extents are extended to page boundaries so that
3150 * dealing with the page cache is a little smoother. */
3151 policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3152 policy->l_extent.end |= ~CFS_PAGE_MASK;
3155 * kms is not valid when either object is completely fresh (so that no
3156 * locks are cached), or object was evicted. In the latter case cached
3157 * lock cannot be used, because it would prime inode state with
3158 * potentially stale LVB.
3163 /* Next, search for already existing extent locks that will cover us */
3164 /* If we're trying to read, we also search for an existing PW lock. The
3165 * VFS and page cache already protect us locally, so lots of readers/
3166 * writers can share a single PW lock.
3168 * There are problems with conversion deadlocks, so instead of
3169 * converting a read lock to a write lock, we'll just enqueue a new
3172 * At some point we should cancel the read lock instead of making them
3173 * send us a blocking callback, but there are problems with canceling
3174 * locks out from other users right now, too. */
3175 mode = einfo->ei_mode;
3176 if (einfo->ei_mode == LCK_PR)
3178 mode = ldlm_lock_match(obd->obd_namespace,
3179 *flags | LDLM_FL_LVB_READY, res_id,
3180 einfo->ei_type, policy, mode, lockh, 0);
3182 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
3184 if (matched->l_ast_data == NULL ||
3185 matched->l_ast_data == einfo->ei_cbdata) {
3186 /* addref the lock only if not async requests and PW
3187 * lock is matched whereas we asked for PR. */
3188 if (!rqset && einfo->ei_mode != mode)
3189 ldlm_lock_addref(lockh, LCK_PR);
3190 osc_set_lock_data_with_check(matched, einfo, *flags);
3192 /* I would like to be able to ASSERT here that
3193 * rss <= kms, but I can't, for reasons which
3194 * are explained in lov_enqueue() */
3197 /* We already have a lock, and it's referenced */
3198 (*upcall)(cookie, ELDLM_OK);
3200 /* For async requests, decref the lock. */
3201 if (einfo->ei_mode != mode)
3202 ldlm_lock_decref(lockh, LCK_PW);
3204 ldlm_lock_decref(lockh, einfo->ei_mode);
3205 LDLM_LOCK_PUT(matched);
3208 ldlm_lock_decref(lockh, mode);
3209 LDLM_LOCK_PUT(matched);
3214 CFS_LIST_HEAD(cancels);
3215 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3216 &RQF_LDLM_ENQUEUE_LVB);
3220 rc = ldlm_prep_enqueue_req(exp, req, &cancels, 0);
3224 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
3226 ptlrpc_request_set_replen(req);
3229 /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
3230 *flags &= ~LDLM_FL_BLOCK_GRANTED;
3232 rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
3233 sizeof(*lvb), lustre_swab_ost_lvb, lockh, async);
3236 struct osc_enqueue_args *aa;
3237 CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3238 aa = ptlrpc_req_async_args(req);
3241 aa->oa_flags = flags;
3242 aa->oa_upcall = upcall;
3243 aa->oa_cookie = cookie;
3245 aa->oa_lockh = lockh;
3247 req->rq_interpret_reply =
3248 (ptlrpc_interpterer_t)osc_enqueue_interpret;
3249 if (rqset == PTLRPCD_SET)
3250 ptlrpcd_add_req(req, PSCOPE_OTHER);
3252 ptlrpc_set_add_req(rqset, req);
3253 } else if (intent) {
3254 ptlrpc_req_finished(req);
3259 rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, rc);
3261 ptlrpc_req_finished(req);
3266 static int osc_enqueue(struct obd_export *exp, struct obd_info *oinfo,
3267 struct ldlm_enqueue_info *einfo,
3268 struct ptlrpc_request_set *rqset)
3270 struct ldlm_res_id res_id;
3274 osc_build_res_name(oinfo->oi_md->lsm_object_id,
3275 oinfo->oi_md->lsm_object_gr, &res_id);
3277 rc = osc_enqueue_base(exp, &res_id, &oinfo->oi_flags, &oinfo->oi_policy,
3278 &oinfo->oi_md->lsm_oinfo[0]->loi_lvb,
3279 oinfo->oi_md->lsm_oinfo[0]->loi_kms_valid,
3280 oinfo->oi_cb_up, oinfo, einfo, oinfo->oi_lockh,
3281 rqset, rqset != NULL);
3285 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3286 __u32 type, ldlm_policy_data_t *policy, __u32 mode,
3287 int *flags, void *data, struct lustre_handle *lockh,
3290 struct obd_device *obd = exp->exp_obd;
3291 int lflags = *flags;
3295 if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
3298 /* Filesystem lock extents are extended to page boundaries so that
3299 * dealing with the page cache is a little smoother */
3300 policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3301 policy->l_extent.end |= ~CFS_PAGE_MASK;
3303 /* Next, search for already existing extent locks that will cover us */
3304 /* If we're trying to read, we also search for an existing PW lock. The
3305 * VFS and page cache already protect us locally, so lots of readers/
3306 * writers can share a single PW lock. */
3310 rc = ldlm_lock_match(obd->obd_namespace, lflags | LDLM_FL_LVB_READY,
3311 res_id, type, policy, rc, lockh, unref);
3314 osc_set_data_with_check(lockh, data, lflags);
3315 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
3316 ldlm_lock_addref(lockh, LCK_PR);
3317 ldlm_lock_decref(lockh, LCK_PW);
3324 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
3328 if (unlikely(mode == LCK_GROUP))
3329 ldlm_lock_decref_and_cancel(lockh, mode);
3331 ldlm_lock_decref(lockh, mode);
3336 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
3337 __u32 mode, struct lustre_handle *lockh)
3340 RETURN(osc_cancel_base(lockh, mode));
3343 static int osc_cancel_unused(struct obd_export *exp,
3344 struct lov_stripe_md *lsm, int flags,
3347 struct obd_device *obd = class_exp2obd(exp);
3348 struct ldlm_res_id res_id, *resp = NULL;
3351 resp = osc_build_res_name(lsm->lsm_object_id,
3352 lsm->lsm_object_gr, &res_id);
3355 return ldlm_cli_cancel_unused(obd->obd_namespace, resp, flags, opaque);
3358 static int osc_statfs_interpret(const struct lu_env *env,
3359 struct ptlrpc_request *req,
3360 struct osc_async_args *aa, int rc)
3362 struct obd_statfs *msfs;
3365 if ((rc == -ENOTCONN || rc == -EAGAIN) &&
3366 (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
3372 msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3374 GOTO(out, rc = -EPROTO);
3377 *aa->aa_oi->oi_osfs = *msfs;
3379 rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3383 static int osc_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
3384 __u64 max_age, struct ptlrpc_request_set *rqset)
3386 struct ptlrpc_request *req;
3387 struct osc_async_args *aa;
3391 /* We could possibly pass max_age in the request (as an absolute
3392 * timestamp or a "seconds.usec ago") so the target can avoid doing
3393 * extra calls into the filesystem if that isn't necessary (e.g.
3394 * during mount that would help a bit). Having relative timestamps
3395 * is not so great if request processing is slow, while absolute
3396 * timestamps are not ideal because they need time synchronization. */
3397 req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3401 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3403 ptlrpc_request_free(req);
3406 ptlrpc_request_set_replen(req);
3407 req->rq_request_portal = OST_CREATE_PORTAL;
3408 ptlrpc_at_set_req_timeout(req);
3410 if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3411 /* procfs requests not want stat in wait for avoid deadlock */
3412 req->rq_no_resend = 1;
3413 req->rq_no_delay = 1;
3416 req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
3417 CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3418 aa = ptlrpc_req_async_args(req);
3421 ptlrpc_set_add_req(rqset, req);
3425 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3426 __u64 max_age, __u32 flags)
3428 struct obd_statfs *msfs;
3429 struct ptlrpc_request *req;
3430 struct obd_import *imp = NULL;
3434 /*Since the request might also come from lprocfs, so we need
3435 *sync this with client_disconnect_export Bug15684*/
3436 down_read(&obd->u.cli.cl_sem);
3437 if (obd->u.cli.cl_import)
3438 imp = class_import_get(obd->u.cli.cl_import);
3439 up_read(&obd->u.cli.cl_sem);
3443 /* We could possibly pass max_age in the request (as an absolute
3444 * timestamp or a "seconds.usec ago") so the target can avoid doing
3445 * extra calls into the filesystem if that isn't necessary (e.g.
3446 * during mount that would help a bit). Having relative timestamps
3447 * is not so great if request processing is slow, while absolute
3448 * timestamps are not ideal because they need time synchronization. */
3449 req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
3451 class_import_put(imp);
3456 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3458 ptlrpc_request_free(req);
3461 ptlrpc_request_set_replen(req);
3462 req->rq_request_portal = OST_CREATE_PORTAL;
3463 ptlrpc_at_set_req_timeout(req);
3465 if (flags & OBD_STATFS_NODELAY) {
3466 /* procfs requests not want stat in wait for avoid deadlock */
3467 req->rq_no_resend = 1;
3468 req->rq_no_delay = 1;
3471 rc = ptlrpc_queue_wait(req);
3475 msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3477 GOTO(out, rc = -EPROTO);
3484 ptlrpc_req_finished(req);
3488 /* Retrieve object striping information.
3490 * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
3491 * the maximum number of OST indices which will fit in the user buffer.
3492 * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
3494 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
3496 /* we use lov_user_md_v3 because it is larger than lov_user_md_v1 */
3497 struct lov_user_md_v3 lum, *lumk;
3498 struct lov_user_ost_data_v1 *lmm_objects;
3499 int rc = 0, lum_size;
3505 /* we only need the header part from user space to get lmm_magic and
3506 * lmm_stripe_count, (the header part is common to v1 and v3) */
3507 lum_size = sizeof(struct lov_user_md_v1);
3508 if (copy_from_user(&lum, lump, lum_size))
3511 if ((lum.lmm_magic != LOV_USER_MAGIC_V1) &&
3512 (lum.lmm_magic != LOV_USER_MAGIC_V3))
3515 /* lov_user_md_vX and lov_mds_md_vX must have the same size */
3516 LASSERT(sizeof(struct lov_user_md_v1) == sizeof(struct lov_mds_md_v1));
3517 LASSERT(sizeof(struct lov_user_md_v3) == sizeof(struct lov_mds_md_v3));
3518 LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lumk->lmm_objects[0]));
3520 /* we can use lov_mds_md_size() to compute lum_size
3521 * because lov_user_md_vX and lov_mds_md_vX have the same size */
3522 if (lum.lmm_stripe_count > 0) {
3523 lum_size = lov_mds_md_size(lum.lmm_stripe_count, lum.lmm_magic);
3524 OBD_ALLOC(lumk, lum_size);
3528 if (lum.lmm_magic == LOV_USER_MAGIC_V1)
3529 lmm_objects = &(((struct lov_user_md_v1 *)lumk)->lmm_objects[0]);
3531 lmm_objects = &(lumk->lmm_objects[0]);
3532 lmm_objects->l_object_id = lsm->lsm_object_id;
3534 lum_size = lov_mds_md_size(0, lum.lmm_magic);
3538 lumk->lmm_object_id = lsm->lsm_object_id;
3539 lumk->lmm_object_gr = lsm->lsm_object_gr;
3540 lumk->lmm_stripe_count = 1;
3542 if (copy_to_user(lump, lumk, lum_size))
3546 OBD_FREE(lumk, lum_size);
3552 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
3553 void *karg, void *uarg)
3555 struct obd_device *obd = exp->exp_obd;
3556 struct obd_ioctl_data *data = karg;
3560 if (!try_module_get(THIS_MODULE)) {
3561 CERROR("Can't get module. Is it alive?");
3565 case OBD_IOC_LOV_GET_CONFIG: {
3567 struct lov_desc *desc;
3568 struct obd_uuid uuid;
3572 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
3573 GOTO(out, err = -EINVAL);
3575 data = (struct obd_ioctl_data *)buf;
3577 if (sizeof(*desc) > data->ioc_inllen1) {
3578 obd_ioctl_freedata(buf, len);
3579 GOTO(out, err = -EINVAL);
3582 if (data->ioc_inllen2 < sizeof(uuid)) {
3583 obd_ioctl_freedata(buf, len);
3584 GOTO(out, err = -EINVAL);
3587 desc = (struct lov_desc *)data->ioc_inlbuf1;
3588 desc->ld_tgt_count = 1;
3589 desc->ld_active_tgt_count = 1;
3590 desc->ld_default_stripe_count = 1;
3591 desc->ld_default_stripe_size = 0;
3592 desc->ld_default_stripe_offset = 0;
3593 desc->ld_pattern = 0;
3594 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
3596 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
3598 err = copy_to_user((void *)uarg, buf, len);
3601 obd_ioctl_freedata(buf, len);
3604 case LL_IOC_LOV_SETSTRIPE:
3605 err = obd_alloc_memmd(exp, karg);
3609 case LL_IOC_LOV_GETSTRIPE:
3610 err = osc_getstripe(karg, uarg);
3612 case OBD_IOC_CLIENT_RECOVER:
3613 err = ptlrpc_recover_import(obd->u.cli.cl_import,
3618 case IOC_OSC_SET_ACTIVE:
3619 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
3622 case OBD_IOC_POLL_QUOTACHECK:
3623 err = lquota_poll_check(quota_interface, exp,
3624 (struct if_quotacheck *)karg);
3626 case OBD_IOC_PING_TARGET:
3627 err = ptlrpc_obd_ping(obd);
3630 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
3631 cmd, cfs_curproc_comm());
3632 GOTO(out, err = -ENOTTY);
3635 module_put(THIS_MODULE);
3639 static int osc_get_info(struct obd_export *exp, obd_count keylen,
3640 void *key, __u32 *vallen, void *val,
3641 struct lov_stripe_md *lsm)
3644 if (!vallen || !val)
3647 if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
3648 __u32 *stripe = val;
3649 *vallen = sizeof(*stripe);
3652 } else if (KEY_IS(KEY_LAST_ID)) {
3653 struct ptlrpc_request *req;
3658 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3659 &RQF_OST_GET_INFO_LAST_ID);
3663 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3664 RCL_CLIENT, keylen);
3665 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3667 ptlrpc_request_free(req);
3671 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3672 memcpy(tmp, key, keylen);
3674 req->rq_no_delay = req->rq_no_resend = 1;
3675 ptlrpc_request_set_replen(req);
3676 rc = ptlrpc_queue_wait(req);
3680 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
3682 GOTO(out, rc = -EPROTO);
3684 *((obd_id *)val) = *reply;
3686 ptlrpc_req_finished(req);
3688 } else if (KEY_IS(KEY_FIEMAP)) {
3689 struct ptlrpc_request *req;
3690 struct ll_user_fiemap *reply;
3694 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3695 &RQF_OST_GET_INFO_FIEMAP);
3699 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
3700 RCL_CLIENT, keylen);
3701 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3702 RCL_CLIENT, *vallen);
3703 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3704 RCL_SERVER, *vallen);
3706 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3708 ptlrpc_request_free(req);
3712 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
3713 memcpy(tmp, key, keylen);
3714 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3715 memcpy(tmp, val, *vallen);
3717 ptlrpc_request_set_replen(req);
3718 rc = ptlrpc_queue_wait(req);
3722 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3724 GOTO(out1, rc = -EPROTO);
3726 memcpy(val, reply, *vallen);
3728 ptlrpc_req_finished(req);
3736 static int osc_setinfo_mds_connect_import(struct obd_import *imp)
3738 struct llog_ctxt *ctxt;
3742 ctxt = llog_get_context(imp->imp_obd, LLOG_MDS_OST_ORIG_CTXT);
3744 rc = llog_initiator_connect(ctxt);
3745 llog_ctxt_put(ctxt);
3747 /* XXX return an error? skip setting below flags? */
3750 spin_lock(&imp->imp_lock);
3751 imp->imp_server_timeout = 1;
3752 imp->imp_pingable = 1;
3753 spin_unlock(&imp->imp_lock);
3754 CDEBUG(D_RPCTRACE, "pinging OST %s\n", obd2cli_tgt(imp->imp_obd));
3759 static int osc_setinfo_mds_conn_interpret(const struct lu_env *env,
3760 struct ptlrpc_request *req,
3767 RETURN(osc_setinfo_mds_connect_import(req->rq_import));
3770 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
3771 void *key, obd_count vallen, void *val,
3772 struct ptlrpc_request_set *set)
3774 struct ptlrpc_request *req;
3775 struct obd_device *obd = exp->exp_obd;
3776 struct obd_import *imp = class_exp2cliimp(exp);
3781 OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
3783 if (KEY_IS(KEY_NEXT_ID)) {
3784 if (vallen != sizeof(obd_id))
3788 obd->u.cli.cl_oscc.oscc_next_id = *((obd_id*)val) + 1;
3789 CDEBUG(D_HA, "%s: set oscc_next_id = "LPU64"\n",
3790 exp->exp_obd->obd_name,
3791 obd->u.cli.cl_oscc.oscc_next_id);
3796 if (KEY_IS(KEY_UNLINKED)) {
3797 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3798 spin_lock(&oscc->oscc_lock);
3799 oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
3800 spin_unlock(&oscc->oscc_lock);
3804 if (KEY_IS(KEY_INIT_RECOV)) {
3805 if (vallen != sizeof(int))
3807 spin_lock(&imp->imp_lock);
3808 imp->imp_initial_recov = *(int *)val;
3809 spin_unlock(&imp->imp_lock);
3810 CDEBUG(D_HA, "%s: set imp_initial_recov = %d\n",
3811 exp->exp_obd->obd_name,
3812 imp->imp_initial_recov);
3816 if (KEY_IS(KEY_CHECKSUM)) {
3817 if (vallen != sizeof(int))
3819 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
3823 if (KEY_IS(KEY_SPTLRPC_CONF)) {
3824 sptlrpc_conf_client_adapt(obd);
3828 if (KEY_IS(KEY_FLUSH_CTX)) {
3829 sptlrpc_import_flush_my_ctx(imp);
3833 if (!set && !KEY_IS(KEY_GRANT_SHRINK))
3836 /* If OST understood OBD_CONNECT_MDS we don't need to tell it we
3837 * are the MDS again. Just do the local setup. b=16839 */
3838 if (KEY_IS(KEY_MDS_CONN) &&
3839 (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_MDS))
3840 RETURN(osc_setinfo_mds_connect_import(imp));
3842 /* We pass all other commands directly to OST. Since nobody calls osc
3843 methods directly and everybody is supposed to go through LOV, we
3844 assume lov checked invalid values for us.
3845 The only recognised values so far are evict_by_nid and mds_conn.
3846 Even if something bad goes through, we'd get a -EINVAL from OST
3849 if (KEY_IS(KEY_GRANT_SHRINK))
3850 req = ptlrpc_request_alloc(imp, &RQF_OST_SET_GRANT_INFO);
3852 req = ptlrpc_request_alloc(imp, &RQF_OST_SET_INFO);
3857 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3858 RCL_CLIENT, keylen);
3859 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
3860 RCL_CLIENT, vallen);
3861 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
3863 ptlrpc_request_free(req);
3867 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3868 memcpy(tmp, key, keylen);
3869 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
3870 memcpy(tmp, val, vallen);
3872 if (KEY_IS(KEY_MDS_CONN)) {
3873 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3875 oscc->oscc_oa.o_gr = (*(__u32 *)val);
3876 oscc->oscc_oa.o_valid |= OBD_MD_FLGROUP;
3877 LASSERT_MDS_GROUP(oscc->oscc_oa.o_gr);
3878 req->rq_no_delay = req->rq_no_resend = 1;
3879 req->rq_interpret_reply = osc_setinfo_mds_conn_interpret;
3880 } else if (KEY_IS(KEY_GRANT_SHRINK)) {
3881 struct osc_grant_args *aa;
3884 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
3885 aa = ptlrpc_req_async_args(req);
3888 ptlrpc_req_finished(req);
3891 *oa = ((struct ost_body *)val)->oa;
3893 req->rq_interpret_reply = osc_shrink_grant_interpret;
3896 ptlrpc_request_set_replen(req);
3897 if (!KEY_IS(KEY_GRANT_SHRINK)) {
3898 LASSERT(set != NULL);
3899 ptlrpc_set_add_req(set, req);
3900 ptlrpc_check_set(NULL, set);
3902 ptlrpcd_add_req(req, PSCOPE_OTHER);
3908 static struct llog_operations osc_size_repl_logops = {
3909 lop_cancel: llog_obd_repl_cancel
3912 static struct llog_operations osc_mds_ost_orig_logops;
3913 static int osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
3914 struct obd_device *tgt, int count,
3915 struct llog_catid *catid, struct obd_uuid *uuid)
3920 LASSERT(olg == &obd->obd_olg);
3921 spin_lock(&obd->obd_dev_lock);
3922 if (osc_mds_ost_orig_logops.lop_setup != llog_obd_origin_setup) {
3923 osc_mds_ost_orig_logops = llog_lvfs_ops;
3924 osc_mds_ost_orig_logops.lop_setup = llog_obd_origin_setup;
3925 osc_mds_ost_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
3926 osc_mds_ost_orig_logops.lop_add = llog_obd_origin_add;
3927 osc_mds_ost_orig_logops.lop_connect = llog_origin_connect;
3929 spin_unlock(&obd->obd_dev_lock);
3931 rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, tgt, count,
3932 &catid->lci_logid, &osc_mds_ost_orig_logops);
3934 CERROR("failed LLOG_MDS_OST_ORIG_CTXT\n");
3938 rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, tgt, count,
3939 NULL, &osc_size_repl_logops);
3941 struct llog_ctxt *ctxt =
3942 llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
3945 CERROR("failed LLOG_SIZE_REPL_CTXT\n");
3950 CERROR("osc '%s' tgt '%s' cnt %d catid %p rc=%d\n",
3951 obd->obd_name, tgt->obd_name, count, catid, rc);
3952 CERROR("logid "LPX64":0x%x\n",
3953 catid->lci_logid.lgl_oid, catid->lci_logid.lgl_ogen);
3958 static int osc_llog_finish(struct obd_device *obd, int count)
3960 struct llog_ctxt *ctxt;
3961 int rc = 0, rc2 = 0;
3964 ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
3966 rc = llog_cleanup(ctxt);
3968 ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
3970 rc2 = llog_cleanup(ctxt);
3977 static int osc_reconnect(const struct lu_env *env,
3978 struct obd_export *exp, struct obd_device *obd,
3979 struct obd_uuid *cluuid,
3980 struct obd_connect_data *data,
3983 struct client_obd *cli = &obd->u.cli;
3985 if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
3988 client_obd_list_lock(&cli->cl_loi_list_lock);
3989 data->ocd_grant = cli->cl_avail_grant ?:
3990 2 * cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT;
3991 lost_grant = cli->cl_lost_grant;
3992 cli->cl_lost_grant = 0;
3993 client_obd_list_unlock(&cli->cl_loi_list_lock);
3995 CDEBUG(D_CACHE, "request ocd_grant: %d cl_avail_grant: %ld "
3996 "cl_lost_grant: %ld\n", data->ocd_grant,
3997 cli->cl_avail_grant, lost_grant);
3998 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
3999 " ocd_grant: %d\n", data->ocd_connect_flags,
4000 data->ocd_version, data->ocd_grant);
4006 static int osc_disconnect(struct obd_export *exp)
4008 struct obd_device *obd = class_exp2obd(exp);
4009 struct llog_ctxt *ctxt;
4012 ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
4014 if (obd->u.cli.cl_conn_count == 1) {
4015 /* Flush any remaining cancel messages out to the
4017 llog_sync(ctxt, exp);
4019 llog_ctxt_put(ctxt);
4021 CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
4025 rc = client_disconnect_export(exp);
4027 * Initially we put del_shrink_grant before disconnect_export, but it
4028 * causes the following problem if setup (connect) and cleanup
4029 * (disconnect) are tangled together.
4030 * connect p1 disconnect p2
4031 * ptlrpc_connect_import
4032 * ............... class_manual_cleanup
4035 * ptlrpc_connect_interrupt
4037 * add this client to shrink list
4039 * Bang! pinger trigger the shrink.
4040 * So the osc should be disconnected from the shrink list, after we
4041 * are sure the import has been destroyed. BUG18662
4043 if (obd->u.cli.cl_import == NULL)
4044 osc_del_shrink_grant(&obd->u.cli);
4048 static int osc_import_event(struct obd_device *obd,
4049 struct obd_import *imp,
4050 enum obd_import_event event)
4052 struct client_obd *cli;
4056 LASSERT(imp->imp_obd == obd);
4059 case IMP_EVENT_DISCON: {
4060 /* Only do this on the MDS OSC's */
4061 if (imp->imp_server_timeout) {
4062 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4064 spin_lock(&oscc->oscc_lock);
4065 oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
4066 spin_unlock(&oscc->oscc_lock);
4069 client_obd_list_lock(&cli->cl_loi_list_lock);
4070 cli->cl_avail_grant = 0;
4071 cli->cl_lost_grant = 0;
4072 client_obd_list_unlock(&cli->cl_loi_list_lock);
4075 case IMP_EVENT_INACTIVE: {
4076 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
4079 case IMP_EVENT_INVALIDATE: {
4080 struct ldlm_namespace *ns = obd->obd_namespace;
4084 env = cl_env_get(&refcheck);
4088 client_obd_list_lock(&cli->cl_loi_list_lock);
4089 /* all pages go to failing rpcs due to the invalid
4091 osc_check_rpcs(env, cli);
4092 client_obd_list_unlock(&cli->cl_loi_list_lock);
4094 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
4095 cl_env_put(env, &refcheck);
4100 case IMP_EVENT_ACTIVE: {
4101 /* Only do this on the MDS OSC's */
4102 if (imp->imp_server_timeout) {
4103 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4105 spin_lock(&oscc->oscc_lock);
4106 oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
4107 spin_unlock(&oscc->oscc_lock);
4109 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
4112 case IMP_EVENT_OCD: {
4113 struct obd_connect_data *ocd = &imp->imp_connect_data;
4115 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
4116 osc_init_grant(&obd->u.cli, ocd);
4119 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
4120 imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
4122 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
4126 CERROR("Unknown import event %d\n", event);
4132 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
4138 rc = ptlrpcd_addref();
4142 rc = client_obd_setup(obd, lcfg);
4146 struct lprocfs_static_vars lvars = { 0 };
4147 struct client_obd *cli = &obd->u.cli;
4149 lprocfs_osc_init_vars(&lvars);
4150 if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
4151 lproc_osc_attach_seqstat(obd);
4152 sptlrpc_lprocfs_cliobd_attach(obd);
4153 ptlrpc_lprocfs_register_obd(obd);
4157 /* We need to allocate a few requests more, because
4158 brw_interpret tries to create new requests before freeing
4159 previous ones. Ideally we want to have 2x max_rpcs_in_flight
4160 reserved, but I afraid that might be too much wasted RAM
4161 in fact, so 2 is just my guess and still should work. */
4162 cli->cl_import->imp_rq_pool =
4163 ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
4165 ptlrpc_add_rqs_to_pool);
4167 CFS_INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
4168 sema_init(&cli->cl_grant_sem, 1);
4174 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
4180 case OBD_CLEANUP_EARLY: {
4181 struct obd_import *imp;
4182 imp = obd->u.cli.cl_import;
4183 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
4184 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
4185 ptlrpc_deactivate_import(imp);
4186 spin_lock(&imp->imp_lock);
4187 imp->imp_pingable = 0;
4188 spin_unlock(&imp->imp_lock);
4191 case OBD_CLEANUP_EXPORTS: {
4192 /* If we set up but never connected, the
4193 client import will not have been cleaned. */
4194 if (obd->u.cli.cl_import) {
4195 struct obd_import *imp;
4196 down_write(&obd->u.cli.cl_sem);
4197 imp = obd->u.cli.cl_import;
4198 CDEBUG(D_CONFIG, "%s: client import never connected\n",
4200 ptlrpc_invalidate_import(imp);
4201 if (imp->imp_rq_pool) {
4202 ptlrpc_free_rq_pool(imp->imp_rq_pool);
4203 imp->imp_rq_pool = NULL;
4205 class_destroy_import(imp);
4206 up_write(&obd->u.cli.cl_sem);
4207 obd->u.cli.cl_import = NULL;
4209 rc = obd_llog_finish(obd, 0);
4211 CERROR("failed to cleanup llogging subsystems\n");
4218 int osc_cleanup(struct obd_device *obd)
4220 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4224 ptlrpc_lprocfs_unregister_obd(obd);
4225 lprocfs_obd_cleanup(obd);
4227 spin_lock(&oscc->oscc_lock);
4228 oscc->oscc_flags &= ~OSCC_FLAG_RECOVERING;
4229 oscc->oscc_flags |= OSCC_FLAG_EXITING;
4230 spin_unlock(&oscc->oscc_lock);
4232 /* free memory of osc quota cache */
4233 lquota_cleanup(quota_interface, obd);
4235 rc = client_obd_cleanup(obd);
4241 int osc_process_config_base(struct obd_device *obd, struct lustre_cfg *lcfg)
4243 struct lprocfs_static_vars lvars = { 0 };
4246 lprocfs_osc_init_vars(&lvars);
4248 switch (lcfg->lcfg_command) {
4250 rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
4260 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
4262 return osc_process_config_base(obd, buf);
4265 struct obd_ops osc_obd_ops = {
4266 .o_owner = THIS_MODULE,
4267 .o_setup = osc_setup,
4268 .o_precleanup = osc_precleanup,
4269 .o_cleanup = osc_cleanup,
4270 .o_add_conn = client_import_add_conn,
4271 .o_del_conn = client_import_del_conn,
4272 .o_connect = client_connect_import,
4273 .o_reconnect = osc_reconnect,
4274 .o_disconnect = osc_disconnect,
4275 .o_statfs = osc_statfs,
4276 .o_statfs_async = osc_statfs_async,
4277 .o_packmd = osc_packmd,
4278 .o_unpackmd = osc_unpackmd,
4279 .o_precreate = osc_precreate,
4280 .o_create = osc_create,
4281 .o_destroy = osc_destroy,
4282 .o_getattr = osc_getattr,
4283 .o_getattr_async = osc_getattr_async,
4284 .o_setattr = osc_setattr,
4285 .o_setattr_async = osc_setattr_async,
4287 .o_punch = osc_punch,
4289 .o_enqueue = osc_enqueue,
4290 .o_change_cbdata = osc_change_cbdata,
4291 .o_cancel = osc_cancel,
4292 .o_cancel_unused = osc_cancel_unused,
4293 .o_iocontrol = osc_iocontrol,
4294 .o_get_info = osc_get_info,
4295 .o_set_info_async = osc_set_info_async,
4296 .o_import_event = osc_import_event,
4297 .o_llog_init = osc_llog_init,
4298 .o_llog_finish = osc_llog_finish,
4299 .o_process_config = osc_process_config,
4302 extern struct lu_kmem_descr osc_caches[];
4303 extern spinlock_t osc_ast_guard;
4304 extern struct lock_class_key osc_ast_guard_class;
4306 int __init osc_init(void)
4308 struct lprocfs_static_vars lvars = { 0 };
4312 /* print an address of _any_ initialized kernel symbol from this
4313 * module, to allow debugging with gdb that doesn't support data
4314 * symbols from modules.*/
4315 CDEBUG(D_CONSOLE, "Lustre OSC module (%p).\n", &osc_caches);
4317 rc = lu_kmem_init(osc_caches);
4319 lprocfs_osc_init_vars(&lvars);
4321 request_module("lquota");
4322 quota_interface = PORTAL_SYMBOL_GET(osc_quota_interface);
4323 lquota_init(quota_interface);
4324 init_obd_quota_ops(quota_interface, &osc_obd_ops);
4326 rc = class_register_type(&osc_obd_ops, NULL, lvars.module_vars,
4327 LUSTRE_OSC_NAME, &osc_device_type);
4329 if (quota_interface)
4330 PORTAL_SYMBOL_PUT(osc_quota_interface);
4331 lu_kmem_fini(osc_caches);
4335 spin_lock_init(&osc_ast_guard);
4336 lockdep_set_class(&osc_ast_guard, &osc_ast_guard_class);
4342 static void /*__exit*/ osc_exit(void)
4344 lu_device_type_fini(&osc_device_type);
4346 lquota_exit(quota_interface);
4347 if (quota_interface)
4348 PORTAL_SYMBOL_PUT(osc_quota_interface);
4350 class_unregister_type(LUSTRE_OSC_NAME);
4351 lu_kmem_fini(osc_caches);
4354 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
4355 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
4356 MODULE_LICENSE("GPL");
4358 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);