Whamcloud - gitweb
b8914c35b26b963db0f52cbe8f950c0a86cb7ffb
[fs/lustre-release.git] / lustre / osc / osc_request.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
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.
11  *
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).
17  *
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
21  *
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
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  */
36
37 #ifndef EXPORT_SYMTAB
38 # define EXPORT_SYMTAB
39 #endif
40 #define DEBUG_SUBSYSTEM S_OSC
41
42 #include <libcfs/libcfs.h>
43
44 #ifndef __KERNEL__
45 # include <liblustre.h>
46 #endif
47
48 #include <lustre_dlm.h>
49 #include <lustre_net.h>
50 #include <lustre/lustre_user.h>
51 #include <obd_cksum.h>
52 #include <obd_ost.h>
53 #include <obd_lov.h>
54
55 #ifdef  __CYGWIN__
56 # include <ctype.h>
57 #endif
58
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"
65
66 static quota_interface_t *quota_interface = NULL;
67 extern quota_interface_t osc_quota_interface;
68
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);
73
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)
77 {
78         int lmm_size;
79         ENTRY;
80
81         lmm_size = sizeof(**lmmp);
82         if (!lmmp)
83                 RETURN(lmm_size);
84
85         if (*lmmp && !lsm) {
86                 OBD_FREE(*lmmp, lmm_size);
87                 *lmmp = NULL;
88                 RETURN(0);
89         }
90
91         if (!*lmmp) {
92                 OBD_ALLOC(*lmmp, lmm_size);
93                 if (!*lmmp)
94                         RETURN(-ENOMEM);
95         }
96
97         if (lsm) {
98                 LASSERT(lsm->lsm_object_id);
99                 LASSERT_SEQ_IS_MDT(lsm->lsm_object_seq);
100                 (*lmmp)->lmm_object_id = cpu_to_le64(lsm->lsm_object_id);
101                 (*lmmp)->lmm_object_seq = cpu_to_le64(lsm->lsm_object_seq);
102         }
103
104         RETURN(lmm_size);
105 }
106
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)
110 {
111         int lsm_size;
112         ENTRY;
113
114         if (lmm != NULL) {
115                 if (lmm_bytes < sizeof (*lmm)) {
116                         CERROR("lov_mds_md too small: %d, need %d\n",
117                                lmm_bytes, (int)sizeof(*lmm));
118                         RETURN(-EINVAL);
119                 }
120                 /* XXX LOV_MAGIC etc check? */
121
122                 if (lmm->lmm_object_id == 0) {
123                         CERROR("lov_mds_md: zero lmm_object_id\n");
124                         RETURN(-EINVAL);
125                 }
126         }
127
128         lsm_size = lov_stripe_md_size(1);
129         if (lsmp == NULL)
130                 RETURN(lsm_size);
131
132         if (*lsmp != NULL && lmm == NULL) {
133                 OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
134                 OBD_FREE(*lsmp, lsm_size);
135                 *lsmp = NULL;
136                 RETURN(0);
137         }
138
139         if (*lsmp == NULL) {
140                 OBD_ALLOC(*lsmp, lsm_size);
141                 if (*lsmp == NULL)
142                         RETURN(-ENOMEM);
143                 OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
144                 if ((*lsmp)->lsm_oinfo[0] == NULL) {
145                         OBD_FREE(*lsmp, lsm_size);
146                         RETURN(-ENOMEM);
147                 }
148                 loi_init((*lsmp)->lsm_oinfo[0]);
149         }
150
151         if (lmm != NULL) {
152                 /* XXX zero *lsmp? */
153                 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
154                 (*lsmp)->lsm_object_seq = le64_to_cpu (lmm->lmm_object_seq);
155                 LASSERT((*lsmp)->lsm_object_id);
156                 LASSERT_SEQ_IS_MDT((*lsmp)->lsm_object_seq);
157         }
158
159         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
160
161         RETURN(lsm_size);
162 }
163
164 static inline void osc_pack_capa(struct ptlrpc_request *req,
165                                  struct ost_body *body, void *capa)
166 {
167         struct obd_capa *oc = (struct obd_capa *)capa;
168         struct lustre_capa *c;
169
170         if (!capa)
171                 return;
172
173         c = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
174         LASSERT(c);
175         capa_cpy(c, oc);
176         body->oa.o_valid |= OBD_MD_FLOSSCAPA;
177         DEBUG_CAPA(D_SEC, c, "pack");
178 }
179
180 static inline void osc_pack_req_body(struct ptlrpc_request *req,
181                                      struct obd_info *oinfo)
182 {
183         struct ost_body *body;
184
185         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
186         LASSERT(body);
187
188         lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
189         osc_pack_capa(req, body, oinfo->oi_capa);
190 }
191
192 static inline void osc_set_capa_size(struct ptlrpc_request *req,
193                                      const struct req_msg_field *field,
194                                      struct obd_capa *oc)
195 {
196         if (oc == NULL)
197                 req_capsule_set_size(&req->rq_pill, field, RCL_CLIENT, 0);
198         else
199                 /* it is already calculated as sizeof struct obd_capa */
200                 ;
201 }
202
203 static int osc_getattr_interpret(const struct lu_env *env,
204                                  struct ptlrpc_request *req,
205                                  struct osc_async_args *aa, int rc)
206 {
207         struct ost_body *body;
208         ENTRY;
209
210         if (rc != 0)
211                 GOTO(out, rc);
212
213         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
214         if (body) {
215                 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
216                 lustre_get_wire_obdo(aa->aa_oi->oi_oa, &body->oa);
217
218                 /* This should really be sent by the OST */
219                 aa->aa_oi->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
220                 aa->aa_oi->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
221         } else {
222                 CDEBUG(D_INFO, "can't unpack ost_body\n");
223                 rc = -EPROTO;
224                 aa->aa_oi->oi_oa->o_valid = 0;
225         }
226 out:
227         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
228         RETURN(rc);
229 }
230
231 static int osc_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
232                              struct ptlrpc_request_set *set)
233 {
234         struct ptlrpc_request *req;
235         struct osc_async_args *aa;
236         int                    rc;
237         ENTRY;
238
239         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
240         if (req == NULL)
241                 RETURN(-ENOMEM);
242
243         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
244         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
245         if (rc) {
246                 ptlrpc_request_free(req);
247                 RETURN(rc);
248         }
249
250         osc_pack_req_body(req, oinfo);
251
252         ptlrpc_request_set_replen(req);
253         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_getattr_interpret;
254
255         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
256         aa = ptlrpc_req_async_args(req);
257         aa->aa_oi = oinfo;
258
259         ptlrpc_set_add_req(set, req);
260         RETURN(0);
261 }
262
263 static int osc_getattr(struct obd_export *exp, struct obd_info *oinfo)
264 {
265         struct ptlrpc_request *req;
266         struct ost_body       *body;
267         int                    rc;
268         ENTRY;
269
270         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
271         if (req == NULL)
272                 RETURN(-ENOMEM);
273
274         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
275         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
276         if (rc) {
277                 ptlrpc_request_free(req);
278                 RETURN(rc);
279         }
280
281         osc_pack_req_body(req, oinfo);
282
283         ptlrpc_request_set_replen(req);
284
285         rc = ptlrpc_queue_wait(req);
286         if (rc)
287                 GOTO(out, rc);
288
289         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
290         if (body == NULL)
291                 GOTO(out, rc = -EPROTO);
292
293         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
294         lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
295
296         /* This should really be sent by the OST */
297         oinfo->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
298         oinfo->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
299
300         EXIT;
301  out:
302         ptlrpc_req_finished(req);
303         return rc;
304 }
305
306 static int osc_setattr(struct obd_export *exp, struct obd_info *oinfo,
307                        struct obd_trans_info *oti)
308 {
309         struct ptlrpc_request *req;
310         struct ost_body       *body;
311         int                    rc;
312         ENTRY;
313
314         LASSERT(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP);
315
316         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
317         if (req == NULL)
318                 RETURN(-ENOMEM);
319
320         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
321         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
322         if (rc) {
323                 ptlrpc_request_free(req);
324                 RETURN(rc);
325         }
326
327         osc_pack_req_body(req, oinfo);
328
329         ptlrpc_request_set_replen(req);
330
331         rc = ptlrpc_queue_wait(req);
332         if (rc)
333                 GOTO(out, rc);
334
335         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
336         if (body == NULL)
337                 GOTO(out, rc = -EPROTO);
338
339         lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
340
341         EXIT;
342 out:
343         ptlrpc_req_finished(req);
344         RETURN(rc);
345 }
346
347 static int osc_setattr_interpret(const struct lu_env *env,
348                                  struct ptlrpc_request *req,
349                                  struct osc_setattr_args *sa, int rc)
350 {
351         struct ost_body *body;
352         ENTRY;
353
354         if (rc != 0)
355                 GOTO(out, rc);
356
357         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
358         if (body == NULL)
359                 GOTO(out, rc = -EPROTO);
360
361         lustre_get_wire_obdo(sa->sa_oa, &body->oa);
362 out:
363         rc = sa->sa_upcall(sa->sa_cookie, rc);
364         RETURN(rc);
365 }
366
367 int osc_setattr_async_base(struct obd_export *exp, struct obd_info *oinfo,
368                            struct obd_trans_info *oti,
369                            obd_enqueue_update_f upcall, void *cookie,
370                            struct ptlrpc_request_set *rqset)
371 {
372         struct ptlrpc_request   *req;
373         struct osc_setattr_args *sa;
374         int                      rc;
375         ENTRY;
376
377         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
378         if (req == NULL)
379                 RETURN(-ENOMEM);
380
381         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
382         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
383         if (rc) {
384                 ptlrpc_request_free(req);
385                 RETURN(rc);
386         }
387
388         if (oti && oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
389                 oinfo->oi_oa->o_lcookie = *oti->oti_logcookies;
390
391         osc_pack_req_body(req, oinfo);
392
393         ptlrpc_request_set_replen(req);
394
395         /* do mds to ost setattr asynchronously */
396         if (!rqset) {
397                 /* Do not wait for response. */
398                 ptlrpcd_add_req(req, PSCOPE_OTHER);
399         } else {
400                 req->rq_interpret_reply =
401                         (ptlrpc_interpterer_t)osc_setattr_interpret;
402
403                 CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
404                 sa = ptlrpc_req_async_args(req);
405                 sa->sa_oa = oinfo->oi_oa;
406                 sa->sa_upcall = upcall;
407                 sa->sa_cookie = cookie;
408
409                 if (rqset == PTLRPCD_SET)
410                         ptlrpcd_add_req(req, PSCOPE_OTHER);
411                 else
412                         ptlrpc_set_add_req(rqset, req);
413         }
414
415         RETURN(0);
416 }
417
418 static int osc_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
419                              struct obd_trans_info *oti,
420                              struct ptlrpc_request_set *rqset)
421 {
422         return osc_setattr_async_base(exp, oinfo, oti,
423                                       oinfo->oi_cb_up, oinfo, rqset);
424 }
425
426 int osc_real_create(struct obd_export *exp, struct obdo *oa,
427                     struct lov_stripe_md **ea, struct obd_trans_info *oti)
428 {
429         struct ptlrpc_request *req;
430         struct ost_body       *body;
431         struct lov_stripe_md  *lsm;
432         int                    rc;
433         ENTRY;
434
435         LASSERT(oa);
436         LASSERT(ea);
437
438         lsm = *ea;
439         if (!lsm) {
440                 rc = obd_alloc_memmd(exp, &lsm);
441                 if (rc < 0)
442                         RETURN(rc);
443         }
444
445         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
446         if (req == NULL)
447                 GOTO(out, rc = -ENOMEM);
448
449         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
450         if (rc) {
451                 ptlrpc_request_free(req);
452                 GOTO(out, rc);
453         }
454
455         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
456         LASSERT(body);
457         lustre_set_wire_obdo(&body->oa, oa);
458
459         ptlrpc_request_set_replen(req);
460
461         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
462             oa->o_flags == OBD_FL_DELORPHAN) {
463                 DEBUG_REQ(D_HA, req,
464                           "delorphan from OST integration");
465                 /* Don't resend the delorphan req */
466                 req->rq_no_resend = req->rq_no_delay = 1;
467         }
468
469         rc = ptlrpc_queue_wait(req);
470         if (rc)
471                 GOTO(out_req, rc);
472
473         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
474         if (body == NULL)
475                 GOTO(out_req, rc = -EPROTO);
476
477         lustre_get_wire_obdo(oa, &body->oa);
478
479         /* This should really be sent by the OST */
480         oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
481         oa->o_valid |= OBD_MD_FLBLKSZ;
482
483         /* XXX LOV STACKING: the lsm that is passed to us from LOV does not
484          * have valid lsm_oinfo data structs, so don't go touching that.
485          * This needs to be fixed in a big way.
486          */
487         lsm->lsm_object_id = oa->o_id;
488         lsm->lsm_object_seq = oa->o_seq;
489         *ea = lsm;
490
491         if (oti != NULL) {
492                 oti->oti_transno = lustre_msg_get_transno(req->rq_repmsg);
493
494                 if (oa->o_valid & OBD_MD_FLCOOKIE) {
495                         if (!oti->oti_logcookies)
496                                 oti_alloc_cookies(oti, 1);
497                         *oti->oti_logcookies = oa->o_lcookie;
498                 }
499         }
500
501         CDEBUG(D_HA, "transno: "LPD64"\n",
502                lustre_msg_get_transno(req->rq_repmsg));
503 out_req:
504         ptlrpc_req_finished(req);
505 out:
506         if (rc && !*ea)
507                 obd_free_memmd(exp, &lsm);
508         RETURN(rc);
509 }
510
511 int osc_punch_base(struct obd_export *exp, struct obd_info *oinfo,
512                    obd_enqueue_update_f upcall, void *cookie,
513                    struct ptlrpc_request_set *rqset)
514 {
515         struct ptlrpc_request   *req;
516         struct osc_setattr_args *sa;
517         struct ost_body         *body;
518         int                      rc;
519         ENTRY;
520
521         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_PUNCH);
522         if (req == NULL)
523                 RETURN(-ENOMEM);
524
525         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
526         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
527         if (rc) {
528                 ptlrpc_request_free(req);
529                 RETURN(rc);
530         }
531         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
532         ptlrpc_at_set_req_timeout(req);
533
534         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
535         LASSERT(body);
536         lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
537         osc_pack_capa(req, body, oinfo->oi_capa);
538
539         ptlrpc_request_set_replen(req);
540
541
542         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_setattr_interpret;
543         CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
544         sa = ptlrpc_req_async_args(req);
545         sa->sa_oa     = oinfo->oi_oa;
546         sa->sa_upcall = upcall;
547         sa->sa_cookie = cookie;
548         if (rqset == PTLRPCD_SET)
549                 ptlrpcd_add_req(req, PSCOPE_OTHER);
550         else
551                 ptlrpc_set_add_req(rqset, req);
552
553         RETURN(0);
554 }
555
556 static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
557                      struct obd_trans_info *oti,
558                      struct ptlrpc_request_set *rqset)
559 {
560         oinfo->oi_oa->o_size   = oinfo->oi_policy.l_extent.start;
561         oinfo->oi_oa->o_blocks = oinfo->oi_policy.l_extent.end;
562         oinfo->oi_oa->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
563         return osc_punch_base(exp, oinfo,
564                               oinfo->oi_cb_up, oinfo, rqset);
565 }
566
567 static int osc_sync(struct obd_export *exp, struct obdo *oa,
568                     struct lov_stripe_md *md, obd_size start, obd_size end,
569                     void *capa)
570 {
571         struct ptlrpc_request *req;
572         struct ost_body       *body;
573         int                    rc;
574         ENTRY;
575
576         if (!oa) {
577                 CDEBUG(D_INFO, "oa NULL\n");
578                 RETURN(-EINVAL);
579         }
580
581         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
582         if (req == NULL)
583                 RETURN(-ENOMEM);
584
585         osc_set_capa_size(req, &RMF_CAPA1, capa);
586         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
587         if (rc) {
588                 ptlrpc_request_free(req);
589                 RETURN(rc);
590         }
591
592         /* overload the size and blocks fields in the oa with start/end */
593         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
594         LASSERT(body);
595         lustre_set_wire_obdo(&body->oa, oa);
596         body->oa.o_size = start;
597         body->oa.o_blocks = end;
598         body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
599         osc_pack_capa(req, body, capa);
600
601         ptlrpc_request_set_replen(req);
602
603         rc = ptlrpc_queue_wait(req);
604         if (rc)
605                 GOTO(out, rc);
606
607         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
608         if (body == NULL)
609                 GOTO(out, rc = -EPROTO);
610
611         lustre_get_wire_obdo(oa, &body->oa);
612
613         EXIT;
614  out:
615         ptlrpc_req_finished(req);
616         return rc;
617 }
618
619 /* Find and cancel locally locks matched by @mode in the resource found by
620  * @objid. Found locks are added into @cancel list. Returns the amount of
621  * locks added to @cancels list. */
622 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
623                                    cfs_list_t *cancels,
624                                    ldlm_mode_t mode, int lock_flags)
625 {
626         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
627         struct ldlm_res_id res_id;
628         struct ldlm_resource *res;
629         int count;
630         ENTRY;
631
632         osc_build_res_name(oa->o_id, oa->o_seq, &res_id);
633         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
634         if (res == NULL)
635                 RETURN(0);
636
637         LDLM_RESOURCE_ADDREF(res);
638         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
639                                            lock_flags, 0, NULL);
640         LDLM_RESOURCE_DELREF(res);
641         ldlm_resource_putref(res);
642         RETURN(count);
643 }
644
645 static int osc_destroy_interpret(const struct lu_env *env,
646                                  struct ptlrpc_request *req, void *data,
647                                  int rc)
648 {
649         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
650
651         cfs_atomic_dec(&cli->cl_destroy_in_flight);
652         cfs_waitq_signal(&cli->cl_destroy_waitq);
653         return 0;
654 }
655
656 static int osc_can_send_destroy(struct client_obd *cli)
657 {
658         if (cfs_atomic_inc_return(&cli->cl_destroy_in_flight) <=
659             cli->cl_max_rpcs_in_flight) {
660                 /* The destroy request can be sent */
661                 return 1;
662         }
663         if (cfs_atomic_dec_return(&cli->cl_destroy_in_flight) <
664             cli->cl_max_rpcs_in_flight) {
665                 /*
666                  * The counter has been modified between the two atomic
667                  * operations.
668                  */
669                 cfs_waitq_signal(&cli->cl_destroy_waitq);
670         }
671         return 0;
672 }
673
674 /* Destroy requests can be async always on the client, and we don't even really
675  * care about the return code since the client cannot do anything at all about
676  * a destroy failure.
677  * When the MDS is unlinking a filename, it saves the file objects into a
678  * recovery llog, and these object records are cancelled when the OST reports
679  * they were destroyed and sync'd to disk (i.e. transaction committed).
680  * If the client dies, or the OST is down when the object should be destroyed,
681  * the records are not cancelled, and when the OST reconnects to the MDS next,
682  * it will retrieve the llog unlink logs and then sends the log cancellation
683  * cookies to the MDS after committing destroy transactions. */
684 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
685                        struct lov_stripe_md *ea, struct obd_trans_info *oti,
686                        struct obd_export *md_export, void *capa)
687 {
688         struct client_obd     *cli = &exp->exp_obd->u.cli;
689         struct ptlrpc_request *req;
690         struct ost_body       *body;
691         CFS_LIST_HEAD(cancels);
692         int rc, count;
693         ENTRY;
694
695         if (!oa) {
696                 CDEBUG(D_INFO, "oa NULL\n");
697                 RETURN(-EINVAL);
698         }
699
700         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
701                                         LDLM_FL_DISCARD_DATA);
702
703         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
704         if (req == NULL) {
705                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
706                 RETURN(-ENOMEM);
707         }
708
709         osc_set_capa_size(req, &RMF_CAPA1, (struct obd_capa *)capa);
710         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
711                                0, &cancels, count);
712         if (rc) {
713                 ptlrpc_request_free(req);
714                 RETURN(rc);
715         }
716
717         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
718         ptlrpc_at_set_req_timeout(req);
719
720         if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE)
721                 oa->o_lcookie = *oti->oti_logcookies;
722         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
723         LASSERT(body);
724         lustre_set_wire_obdo(&body->oa, oa);
725
726         osc_pack_capa(req, body, (struct obd_capa *)capa);
727         ptlrpc_request_set_replen(req);
728
729         /* don't throttle destroy RPCs for the MDT */
730         if (!(cli->cl_import->imp_connect_flags_orig & OBD_CONNECT_MDS)) {
731                 req->rq_interpret_reply = osc_destroy_interpret;
732                 if (!osc_can_send_destroy(cli)) {
733                         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP,
734                                                           NULL);
735
736                         /*
737                          * Wait until the number of on-going destroy RPCs drops
738                          * under max_rpc_in_flight
739                          */
740                         l_wait_event_exclusive(cli->cl_destroy_waitq,
741                                                osc_can_send_destroy(cli), &lwi);
742                 }
743         }
744
745         /* Do not wait for response */
746         ptlrpcd_add_req(req, PSCOPE_OTHER);
747         RETURN(0);
748 }
749
750 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
751                                 long writing_bytes)
752 {
753         obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
754
755         LASSERT(!(oa->o_valid & bits));
756
757         oa->o_valid |= bits;
758         client_obd_list_lock(&cli->cl_loi_list_lock);
759         oa->o_dirty = cli->cl_dirty;
760         if (cli->cl_dirty - cli->cl_dirty_transit > cli->cl_dirty_max) {
761                 CERROR("dirty %lu - %lu > dirty_max %lu\n",
762                        cli->cl_dirty, cli->cl_dirty_transit, cli->cl_dirty_max);
763                 oa->o_undirty = 0;
764         } else if (cfs_atomic_read(&obd_dirty_pages) -
765                    cfs_atomic_read(&obd_dirty_transit_pages) >
766                    obd_max_dirty_pages + 1){
767                 /* The cfs_atomic_read() allowing the cfs_atomic_inc() are
768                  * not covered by a lock thus they may safely race and trip
769                  * this CERROR() unless we add in a small fudge factor (+1). */
770                 CERROR("dirty %d - %d > system dirty_max %d\n",
771                        cfs_atomic_read(&obd_dirty_pages),
772                        cfs_atomic_read(&obd_dirty_transit_pages),
773                        obd_max_dirty_pages);
774                 oa->o_undirty = 0;
775         } else if (cli->cl_dirty_max - cli->cl_dirty > 0x7fffffff) {
776                 CERROR("dirty %lu - dirty_max %lu too big???\n",
777                        cli->cl_dirty, cli->cl_dirty_max);
778                 oa->o_undirty = 0;
779         } else {
780                 long max_in_flight = (cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT)*
781                                 (cli->cl_max_rpcs_in_flight + 1);
782                 oa->o_undirty = max(cli->cl_dirty_max, max_in_flight);
783         }
784         oa->o_grant = cli->cl_avail_grant;
785         oa->o_dropped = cli->cl_lost_grant;
786         cli->cl_lost_grant = 0;
787         client_obd_list_unlock(&cli->cl_loi_list_lock);
788         CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
789                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
790
791 }
792
793 static void osc_update_next_shrink(struct client_obd *cli)
794 {
795         cli->cl_next_shrink_grant =
796                 cfs_time_shift(cli->cl_grant_shrink_interval);
797         CDEBUG(D_CACHE, "next time %ld to shrink grant \n",
798                cli->cl_next_shrink_grant);
799 }
800
801 /* caller must hold loi_list_lock */
802 static void osc_consume_write_grant(struct client_obd *cli,
803                                     struct brw_page *pga)
804 {
805         LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock.lock);
806         LASSERT(!(pga->flag & OBD_BRW_FROM_GRANT));
807         cfs_atomic_inc(&obd_dirty_pages);
808         cli->cl_dirty += CFS_PAGE_SIZE;
809         cli->cl_avail_grant -= CFS_PAGE_SIZE;
810         pga->flag |= OBD_BRW_FROM_GRANT;
811         CDEBUG(D_CACHE, "using %lu grant credits for brw %p page %p\n",
812                CFS_PAGE_SIZE, pga, pga->pg);
813         LASSERT(cli->cl_avail_grant >= 0);
814         osc_update_next_shrink(cli);
815 }
816
817 /* the companion to osc_consume_write_grant, called when a brw has completed.
818  * must be called with the loi lock held. */
819 static void osc_release_write_grant(struct client_obd *cli,
820                                     struct brw_page *pga, int sent)
821 {
822         int blocksize = cli->cl_import->imp_obd->obd_osfs.os_bsize ? : 4096;
823         ENTRY;
824
825         LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock.lock);
826         if (!(pga->flag & OBD_BRW_FROM_GRANT)) {
827                 EXIT;
828                 return;
829         }
830
831         pga->flag &= ~OBD_BRW_FROM_GRANT;
832         cfs_atomic_dec(&obd_dirty_pages);
833         cli->cl_dirty -= CFS_PAGE_SIZE;
834         if (pga->flag & OBD_BRW_NOCACHE) {
835                 pga->flag &= ~OBD_BRW_NOCACHE;
836                 cfs_atomic_dec(&obd_dirty_transit_pages);
837                 cli->cl_dirty_transit -= CFS_PAGE_SIZE;
838         }
839         if (!sent) {
840                 cli->cl_lost_grant += CFS_PAGE_SIZE;
841                 CDEBUG(D_CACHE, "lost grant: %lu avail grant: %lu dirty: %lu\n",
842                        cli->cl_lost_grant, cli->cl_avail_grant, cli->cl_dirty);
843         } else if (CFS_PAGE_SIZE != blocksize && pga->count != CFS_PAGE_SIZE) {
844                 /* For short writes we shouldn't count parts of pages that
845                  * span a whole block on the OST side, or our accounting goes
846                  * wrong.  Should match the code in filter_grant_check. */
847                 int offset = pga->off & ~CFS_PAGE_MASK;
848                 int count = pga->count + (offset & (blocksize - 1));
849                 int end = (offset + pga->count) & (blocksize - 1);
850                 if (end)
851                         count += blocksize - end;
852
853                 cli->cl_lost_grant += CFS_PAGE_SIZE - count;
854                 CDEBUG(D_CACHE, "lost %lu grant: %lu avail: %lu dirty: %lu\n",
855                        CFS_PAGE_SIZE - count, cli->cl_lost_grant,
856                        cli->cl_avail_grant, cli->cl_dirty);
857         }
858
859         EXIT;
860 }
861
862 static unsigned long rpcs_in_flight(struct client_obd *cli)
863 {
864         return cli->cl_r_in_flight + cli->cl_w_in_flight;
865 }
866
867 /* caller must hold loi_list_lock */
868 void osc_wake_cache_waiters(struct client_obd *cli)
869 {
870         cfs_list_t *l, *tmp;
871         struct osc_cache_waiter *ocw;
872
873         ENTRY;
874         cfs_list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
875                 /* if we can't dirty more, we must wait until some is written */
876                 if ((cli->cl_dirty + CFS_PAGE_SIZE > cli->cl_dirty_max) ||
877                    (cfs_atomic_read(&obd_dirty_pages) + 1 >
878                     obd_max_dirty_pages)) {
879                         CDEBUG(D_CACHE, "no dirty room: dirty: %ld "
880                                "osc max %ld, sys max %d\n", cli->cl_dirty,
881                                cli->cl_dirty_max, obd_max_dirty_pages);
882                         return;
883                 }
884
885                 /* if still dirty cache but no grant wait for pending RPCs that
886                  * may yet return us some grant before doing sync writes */
887                 if (cli->cl_w_in_flight && cli->cl_avail_grant < CFS_PAGE_SIZE) {
888                         CDEBUG(D_CACHE, "%u BRW writes in flight, no grant\n",
889                                cli->cl_w_in_flight);
890                         return;
891                 }
892
893                 ocw = cfs_list_entry(l, struct osc_cache_waiter, ocw_entry);
894                 cfs_list_del_init(&ocw->ocw_entry);
895                 if (cli->cl_avail_grant < CFS_PAGE_SIZE) {
896                         /* no more RPCs in flight to return grant, do sync IO */
897                         ocw->ocw_rc = -EDQUOT;
898                         CDEBUG(D_INODE, "wake oap %p for sync\n", ocw->ocw_oap);
899                 } else {
900                         osc_consume_write_grant(cli,
901                                                 &ocw->ocw_oap->oap_brw_page);
902                 }
903
904                 cfs_waitq_signal(&ocw->ocw_waitq);
905         }
906
907         EXIT;
908 }
909
910 static void __osc_update_grant(struct client_obd *cli, obd_size grant)
911 {
912         client_obd_list_lock(&cli->cl_loi_list_lock);
913         cli->cl_avail_grant += grant;
914         client_obd_list_unlock(&cli->cl_loi_list_lock);
915 }
916
917 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
918 {
919         if (body->oa.o_valid & OBD_MD_FLGRANT) {
920                 CDEBUG(D_CACHE, "got "LPU64" extra grant\n", body->oa.o_grant);
921                 __osc_update_grant(cli, body->oa.o_grant);
922         }
923 }
924
925 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
926                               void *key, obd_count vallen, void *val,
927                               struct ptlrpc_request_set *set);
928
929 static int osc_shrink_grant_interpret(const struct lu_env *env,
930                                       struct ptlrpc_request *req,
931                                       void *aa, int rc)
932 {
933         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
934         struct obdo *oa = ((struct osc_grant_args *)aa)->aa_oa;
935         struct ost_body *body;
936
937         if (rc != 0) {
938                 __osc_update_grant(cli, oa->o_grant);
939                 GOTO(out, rc);
940         }
941
942         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
943         LASSERT(body);
944         osc_update_grant(cli, body);
945 out:
946         OBD_FREE_PTR(oa);
947         return rc;
948 }
949
950 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
951 {
952         client_obd_list_lock(&cli->cl_loi_list_lock);
953         oa->o_grant = cli->cl_avail_grant / 4;
954         cli->cl_avail_grant -= oa->o_grant;
955         client_obd_list_unlock(&cli->cl_loi_list_lock);
956         oa->o_flags |= OBD_FL_SHRINK_GRANT;
957         osc_update_next_shrink(cli);
958 }
959
960 /* Shrink the current grant, either from some large amount to enough for a
961  * full set of in-flight RPCs, or if we have already shrunk to that limit
962  * then to enough for a single RPC.  This avoids keeping more grant than
963  * needed, and avoids shrinking the grant piecemeal. */
964 static int osc_shrink_grant(struct client_obd *cli)
965 {
966         long target = (cli->cl_max_rpcs_in_flight + 1) *
967                       cli->cl_max_pages_per_rpc;
968
969         client_obd_list_lock(&cli->cl_loi_list_lock);
970         if (cli->cl_avail_grant <= target)
971                 target = cli->cl_max_pages_per_rpc;
972         client_obd_list_unlock(&cli->cl_loi_list_lock);
973
974         return osc_shrink_grant_to_target(cli, target);
975 }
976
977 int osc_shrink_grant_to_target(struct client_obd *cli, long target)
978 {
979         int    rc = 0;
980         struct ost_body     *body;
981         ENTRY;
982
983         client_obd_list_lock(&cli->cl_loi_list_lock);
984         /* Don't shrink if we are already above or below the desired limit
985          * We don't want to shrink below a single RPC, as that will negatively
986          * impact block allocation and long-term performance. */
987         if (target < cli->cl_max_pages_per_rpc)
988                 target = cli->cl_max_pages_per_rpc;
989
990         if (target >= cli->cl_avail_grant) {
991                 client_obd_list_unlock(&cli->cl_loi_list_lock);
992                 RETURN(0);
993         }
994         client_obd_list_unlock(&cli->cl_loi_list_lock);
995
996         OBD_ALLOC_PTR(body);
997         if (!body)
998                 RETURN(-ENOMEM);
999
1000         osc_announce_cached(cli, &body->oa, 0);
1001
1002         client_obd_list_lock(&cli->cl_loi_list_lock);
1003         body->oa.o_grant = cli->cl_avail_grant - target;
1004         cli->cl_avail_grant = target;
1005         client_obd_list_unlock(&cli->cl_loi_list_lock);
1006         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
1007         osc_update_next_shrink(cli);
1008
1009         rc = osc_set_info_async(cli->cl_import->imp_obd->obd_self_export,
1010                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
1011                                 sizeof(*body), body, NULL);
1012         if (rc != 0)
1013                 __osc_update_grant(cli, body->oa.o_grant);
1014         OBD_FREE_PTR(body);
1015         RETURN(rc);
1016 }
1017
1018 #define GRANT_SHRINK_LIMIT PTLRPC_MAX_BRW_SIZE
1019 static int osc_should_shrink_grant(struct client_obd *client)
1020 {
1021         cfs_time_t time = cfs_time_current();
1022         cfs_time_t next_shrink = client->cl_next_shrink_grant;
1023
1024         if ((client->cl_import->imp_connect_data.ocd_connect_flags &
1025              OBD_CONNECT_GRANT_SHRINK) == 0)
1026                 return 0;
1027
1028         if (cfs_time_aftereq(time, next_shrink - 5 * CFS_TICK)) {
1029                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
1030                     client->cl_avail_grant > GRANT_SHRINK_LIMIT)
1031                         return 1;
1032                 else
1033                         osc_update_next_shrink(client);
1034         }
1035         return 0;
1036 }
1037
1038 static int osc_grant_shrink_grant_cb(struct timeout_item *item, void *data)
1039 {
1040         struct client_obd *client;
1041
1042         cfs_list_for_each_entry(client, &item->ti_obd_list,
1043                                 cl_grant_shrink_list) {
1044                 if (osc_should_shrink_grant(client))
1045                         osc_shrink_grant(client);
1046         }
1047         return 0;
1048 }
1049
1050 static int osc_add_shrink_grant(struct client_obd *client)
1051 {
1052         int rc;
1053
1054         rc = ptlrpc_add_timeout_client(client->cl_grant_shrink_interval,
1055                                        TIMEOUT_GRANT,
1056                                        osc_grant_shrink_grant_cb, NULL,
1057                                        &client->cl_grant_shrink_list);
1058         if (rc) {
1059                 CERROR("add grant client %s error %d\n",
1060                         client->cl_import->imp_obd->obd_name, rc);
1061                 return rc;
1062         }
1063         CDEBUG(D_CACHE, "add grant client %s \n",
1064                client->cl_import->imp_obd->obd_name);
1065         osc_update_next_shrink(client);
1066         return 0;
1067 }
1068
1069 static int osc_del_shrink_grant(struct client_obd *client)
1070 {
1071         return ptlrpc_del_timeout_client(&client->cl_grant_shrink_list,
1072                                          TIMEOUT_GRANT);
1073 }
1074
1075 static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1076 {
1077         /*
1078          * ocd_grant is the total grant amount we're expect to hold: if we've
1079          * been evicted, it's the new avail_grant amount, cl_dirty will drop
1080          * to 0 as inflight RPCs fail out; otherwise, it's avail_grant + dirty.
1081          *
1082          * race is tolerable here: if we're evicted, but imp_state already
1083          * left EVICTED state, then cl_dirty must be 0 already.
1084          */
1085         client_obd_list_lock(&cli->cl_loi_list_lock);
1086         if (cli->cl_import->imp_state == LUSTRE_IMP_EVICTED)
1087                 cli->cl_avail_grant = ocd->ocd_grant;
1088         else
1089                 cli->cl_avail_grant = ocd->ocd_grant - cli->cl_dirty;
1090         client_obd_list_unlock(&cli->cl_loi_list_lock);
1091
1092         CDEBUG(D_CACHE, "setting cl_avail_grant: %ld cl_lost_grant: %ld \n",
1093                cli->cl_avail_grant, cli->cl_lost_grant);
1094         LASSERT(cli->cl_avail_grant >= 0);
1095
1096         if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1097             cfs_list_empty(&cli->cl_grant_shrink_list))
1098                 osc_add_shrink_grant(cli);
1099 }
1100
1101 /* We assume that the reason this OSC got a short read is because it read
1102  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1103  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1104  * this stripe never got written at or beyond this stripe offset yet. */
1105 static void handle_short_read(int nob_read, obd_count page_count,
1106                               struct brw_page **pga)
1107 {
1108         char *ptr;
1109         int i = 0;
1110
1111         /* skip bytes read OK */
1112         while (nob_read > 0) {
1113                 LASSERT (page_count > 0);
1114
1115                 if (pga[i]->count > nob_read) {
1116                         /* EOF inside this page */
1117                         ptr = cfs_kmap(pga[i]->pg) +
1118                                 (pga[i]->off & ~CFS_PAGE_MASK);
1119                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1120                         cfs_kunmap(pga[i]->pg);
1121                         page_count--;
1122                         i++;
1123                         break;
1124                 }
1125
1126                 nob_read -= pga[i]->count;
1127                 page_count--;
1128                 i++;
1129         }
1130
1131         /* zero remaining pages */
1132         while (page_count-- > 0) {
1133                 ptr = cfs_kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1134                 memset(ptr, 0, pga[i]->count);
1135                 cfs_kunmap(pga[i]->pg);
1136                 i++;
1137         }
1138 }
1139
1140 static int check_write_rcs(struct ptlrpc_request *req,
1141                            int requested_nob, int niocount,
1142                            obd_count page_count, struct brw_page **pga)
1143 {
1144         int     i;
1145         __u32   *remote_rcs;
1146
1147         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1148                                                   sizeof(*remote_rcs) *
1149                                                   niocount);
1150         if (remote_rcs == NULL) {
1151                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1152                 return(-EPROTO);
1153         }
1154
1155         /* return error if any niobuf was in error */
1156         for (i = 0; i < niocount; i++) {
1157                 if (remote_rcs[i] < 0)
1158                         return(remote_rcs[i]);
1159
1160                 if (remote_rcs[i] != 0) {
1161                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1162                                 i, remote_rcs[i], req);
1163                         return(-EPROTO);
1164                 }
1165         }
1166
1167         if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1168                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1169                        req->rq_bulk->bd_nob_transferred, requested_nob);
1170                 return(-EPROTO);
1171         }
1172
1173         return (0);
1174 }
1175
1176 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1177 {
1178         if (p1->flag != p2->flag) {
1179                 unsigned mask = ~(OBD_BRW_FROM_GRANT|
1180                                   OBD_BRW_NOCACHE|OBD_BRW_SYNC|OBD_BRW_ASYNC);
1181
1182                 /* warn if we try to combine flags that we don't know to be
1183                  * safe to combine */
1184                 if ((p1->flag & mask) != (p2->flag & mask))
1185                         CERROR("is it ok to have flags 0x%x and 0x%x in the "
1186                                "same brw?\n", p1->flag, p2->flag);
1187                 return 0;
1188         }
1189
1190         return (p1->off + p1->count == p2->off);
1191 }
1192
1193 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1194                                    struct brw_page **pga, int opc,
1195                                    cksum_type_t cksum_type)
1196 {
1197         __u32 cksum;
1198         int i = 0;
1199
1200         LASSERT (pg_count > 0);
1201         cksum = init_checksum(cksum_type);
1202         while (nob > 0 && pg_count > 0) {
1203                 unsigned char *ptr = cfs_kmap(pga[i]->pg);
1204                 int off = pga[i]->off & ~CFS_PAGE_MASK;
1205                 int count = pga[i]->count > nob ? nob : pga[i]->count;
1206
1207                 /* corrupt the data before we compute the checksum, to
1208                  * simulate an OST->client data error */
1209                 if (i == 0 && opc == OST_READ &&
1210                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))
1211                         memcpy(ptr + off, "bad1", min(4, nob));
1212                 cksum = compute_checksum(cksum, ptr + off, count, cksum_type);
1213                 cfs_kunmap(pga[i]->pg);
1214                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d checksum %x\n",
1215                                off, cksum);
1216
1217                 nob -= pga[i]->count;
1218                 pg_count--;
1219                 i++;
1220         }
1221         /* For sending we only compute the wrong checksum instead
1222          * of corrupting the data so it is still correct on a redo */
1223         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1224                 cksum++;
1225
1226         return cksum;
1227 }
1228
1229 static int osc_brw_prep_request(int cmd, struct client_obd *cli,struct obdo *oa,
1230                                 struct lov_stripe_md *lsm, obd_count page_count,
1231                                 struct brw_page **pga,
1232                                 struct ptlrpc_request **reqp,
1233                                 struct obd_capa *ocapa, int reserve)
1234 {
1235         struct ptlrpc_request   *req;
1236         struct ptlrpc_bulk_desc *desc;
1237         struct ost_body         *body;
1238         struct obd_ioobj        *ioobj;
1239         struct niobuf_remote    *niobuf;
1240         int niocount, i, requested_nob, opc, rc;
1241         struct osc_brw_async_args *aa;
1242         struct req_capsule      *pill;
1243         struct brw_page *pg_prev;
1244
1245         ENTRY;
1246         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1247                 RETURN(-ENOMEM); /* Recoverable */
1248         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1249                 RETURN(-EINVAL); /* Fatal */
1250
1251         if ((cmd & OBD_BRW_WRITE) != 0) {
1252                 opc = OST_WRITE;
1253                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1254                                                 cli->cl_import->imp_rq_pool,
1255                                                 &RQF_OST_BRW_WRITE);
1256         } else {
1257                 opc = OST_READ;
1258                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW_READ);
1259         }
1260         if (req == NULL)
1261                 RETURN(-ENOMEM);
1262
1263         for (niocount = i = 1; i < page_count; i++) {
1264                 if (!can_merge_pages(pga[i - 1], pga[i]))
1265                         niocount++;
1266         }
1267
1268         pill = &req->rq_pill;
1269         req_capsule_set_size(pill, &RMF_OBD_IOOBJ, RCL_CLIENT,
1270                              sizeof(*ioobj));
1271         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1272                              niocount * sizeof(*niobuf));
1273         osc_set_capa_size(req, &RMF_CAPA1, ocapa);
1274
1275         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1276         if (rc) {
1277                 ptlrpc_request_free(req);
1278                 RETURN(rc);
1279         }
1280         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1281         ptlrpc_at_set_req_timeout(req);
1282
1283         if (opc == OST_WRITE)
1284                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1285                                             BULK_GET_SOURCE, OST_BULK_PORTAL);
1286         else
1287                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1288                                             BULK_PUT_SINK, OST_BULK_PORTAL);
1289
1290         if (desc == NULL)
1291                 GOTO(out, rc = -ENOMEM);
1292         /* NB request now owns desc and will free it when it gets freed */
1293
1294         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1295         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1296         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1297         LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
1298
1299         lustre_set_wire_obdo(&body->oa, oa);
1300
1301         obdo_to_ioobj(oa, ioobj);
1302         ioobj->ioo_bufcnt = niocount;
1303         osc_pack_capa(req, body, ocapa);
1304         LASSERT (page_count > 0);
1305         pg_prev = pga[0];
1306         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1307                 struct brw_page *pg = pga[i];
1308
1309                 LASSERT(pg->count > 0);
1310                 LASSERTF((pg->off & ~CFS_PAGE_MASK) + pg->count <= CFS_PAGE_SIZE,
1311                          "i: %d pg: %p off: "LPU64", count: %u\n", i, pg,
1312                          pg->off, pg->count);
1313 #ifdef __linux__
1314                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1315                          "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
1316                          " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
1317                          i, page_count,
1318                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1319                          pg_prev->pg, page_private(pg_prev->pg),
1320                          pg_prev->pg->index, pg_prev->off);
1321 #else
1322                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1323                          "i %d p_c %u\n", i, page_count);
1324 #endif
1325                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1326                         (pg->flag & OBD_BRW_SRVLOCK));
1327
1328                 ptlrpc_prep_bulk_page(desc, pg->pg, pg->off & ~CFS_PAGE_MASK,
1329                                       pg->count);
1330                 requested_nob += pg->count;
1331
1332                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1333                         niobuf--;
1334                         niobuf->len += pg->count;
1335                 } else {
1336                         niobuf->offset = pg->off;
1337                         niobuf->len    = pg->count;
1338                         niobuf->flags  = pg->flag;
1339                 }
1340                 pg_prev = pg;
1341         }
1342
1343         LASSERTF((void *)(niobuf - niocount) ==
1344                 req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE),
1345                 "want %p - real %p\n", req_capsule_client_get(&req->rq_pill,
1346                 &RMF_NIOBUF_REMOTE), (void *)(niobuf - niocount));
1347
1348         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1349         if (osc_should_shrink_grant(cli))
1350                 osc_shrink_grant_local(cli, &body->oa);
1351
1352         /* size[REQ_REC_OFF] still sizeof (*body) */
1353         if (opc == OST_WRITE) {
1354                 if (unlikely(cli->cl_checksum) &&
1355                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1356                         /* store cl_cksum_type in a local variable since
1357                          * it can be changed via lprocfs */
1358                         cksum_type_t cksum_type = cli->cl_cksum_type;
1359
1360                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1361                                 oa->o_flags &= OBD_FL_LOCAL_MASK;
1362                                 body->oa.o_flags = 0;
1363                         }
1364                         body->oa.o_flags |= cksum_type_pack(cksum_type);
1365                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1366                         body->oa.o_cksum = osc_checksum_bulk(requested_nob,
1367                                                              page_count, pga,
1368                                                              OST_WRITE,
1369                                                              cksum_type);
1370                         CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1371                                body->oa.o_cksum);
1372                         /* save this in 'oa', too, for later checking */
1373                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1374                         oa->o_flags |= cksum_type_pack(cksum_type);
1375                 } else {
1376                         /* clear out the checksum flag, in case this is a
1377                          * resend but cl_checksum is no longer set. b=11238 */
1378                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1379                 }
1380                 oa->o_cksum = body->oa.o_cksum;
1381                 /* 1 RC per niobuf */
1382                 req_capsule_set_size(pill, &RMF_RCS, RCL_SERVER,
1383                                      sizeof(__u32) * niocount);
1384         } else {
1385                 if (unlikely(cli->cl_checksum) &&
1386                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1387                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1388                                 body->oa.o_flags = 0;
1389                         body->oa.o_flags |= cksum_type_pack(cli->cl_cksum_type);
1390                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1391                 }
1392         }
1393         ptlrpc_request_set_replen(req);
1394
1395         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1396         aa = ptlrpc_req_async_args(req);
1397         aa->aa_oa = oa;
1398         aa->aa_requested_nob = requested_nob;
1399         aa->aa_nio_count = niocount;
1400         aa->aa_page_count = page_count;
1401         aa->aa_resends = 0;
1402         aa->aa_ppga = pga;
1403         aa->aa_cli = cli;
1404         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1405         if (ocapa && reserve)
1406                 aa->aa_ocapa = capa_get(ocapa);
1407
1408         *reqp = req;
1409         RETURN(0);
1410
1411  out:
1412         ptlrpc_req_finished(req);
1413         RETURN(rc);
1414 }
1415
1416 static int check_write_checksum(struct obdo *oa, const lnet_process_id_t *peer,
1417                                 __u32 client_cksum, __u32 server_cksum, int nob,
1418                                 obd_count page_count, struct brw_page **pga,
1419                                 cksum_type_t client_cksum_type)
1420 {
1421         __u32 new_cksum;
1422         char *msg;
1423         cksum_type_t cksum_type;
1424
1425         if (server_cksum == client_cksum) {
1426                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1427                 return 0;
1428         }
1429
1430         /* If this is mmaped file - it can be changed at any time */
1431         if (oa->o_valid & OBD_MD_FLFLAGS && oa->o_flags & OBD_FL_MMAP)
1432                 return 1;
1433
1434         if (oa->o_valid & OBD_MD_FLFLAGS)
1435                 cksum_type = cksum_type_unpack(oa->o_flags);
1436         else
1437                 cksum_type = OBD_CKSUM_CRC32;
1438
1439         new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1440                                       cksum_type);
1441
1442         if (cksum_type != client_cksum_type)
1443                 msg = "the server did not use the checksum type specified in "
1444                       "the original request - likely a protocol problem";
1445         else if (new_cksum == server_cksum)
1446                 msg = "changed on the client after we checksummed it - "
1447                       "likely false positive due to mmap IO (bug 11742)";
1448         else if (new_cksum == client_cksum)
1449                 msg = "changed in transit before arrival at OST";
1450         else
1451                 msg = "changed in transit AND doesn't match the original - "
1452                       "likely false positive due to mmap IO (bug 11742)";
1453
1454         LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inode "DFID
1455                            " object "LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1456                            msg, libcfs_nid2str(peer->nid),
1457                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1458                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1459                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1460                            oa->o_id,
1461                            oa->o_valid & OBD_MD_FLGROUP ? oa->o_seq : (__u64)0,
1462                            pga[0]->off,
1463                            pga[page_count-1]->off + pga[page_count-1]->count - 1);
1464         CERROR("original client csum %x (type %x), server csum %x (type %x), "
1465                "client csum now %x\n", client_cksum, client_cksum_type,
1466                server_cksum, cksum_type, new_cksum);
1467         return 1;
1468 }
1469
1470 /* Note rc enters this function as number of bytes transferred */
1471 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1472 {
1473         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1474         const lnet_process_id_t *peer =
1475                         &req->rq_import->imp_connection->c_peer;
1476         struct client_obd *cli = aa->aa_cli;
1477         struct ost_body *body;
1478         __u32 client_cksum = 0;
1479         ENTRY;
1480
1481         if (rc < 0 && rc != -EDQUOT) {
1482                 DEBUG_REQ(D_INFO, req, "Failed request with rc = %d\n", rc);
1483                 RETURN(rc);
1484         }
1485
1486         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1487         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1488         if (body == NULL) {
1489                 DEBUG_REQ(D_INFO, req, "Can't unpack body\n");
1490                 RETURN(-EPROTO);
1491         }
1492
1493 #ifdef HAVE_QUOTA_SUPPORT
1494         /* set/clear over quota flag for a uid/gid */
1495         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1496             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
1497                 unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
1498
1499                 CDEBUG(D_QUOTA, "setdq for [%u %u] with valid "LPX64", flags %x\n",
1500                        body->oa.o_uid, body->oa.o_gid, body->oa.o_valid,
1501                        body->oa.o_flags);
1502                 lquota_setdq(quota_interface, cli, qid, body->oa.o_valid,
1503                              body->oa.o_flags);
1504         }
1505 #endif
1506
1507         osc_update_grant(cli, body);
1508
1509         if (rc < 0)
1510                 RETURN(rc);
1511
1512         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1513                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1514
1515         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1516                 if (rc > 0) {
1517                         CERROR("Unexpected +ve rc %d\n", rc);
1518                         RETURN(-EPROTO);
1519                 }
1520                 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1521
1522                 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1523                         RETURN(-EAGAIN);
1524
1525                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1526                     check_write_checksum(&body->oa, peer, client_cksum,
1527                                          body->oa.o_cksum, aa->aa_requested_nob,
1528                                          aa->aa_page_count, aa->aa_ppga,
1529                                          cksum_type_unpack(aa->aa_oa->o_flags)))
1530                         RETURN(-EAGAIN);
1531
1532                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1533                                      aa->aa_page_count, aa->aa_ppga);
1534                 GOTO(out, rc);
1535         }
1536
1537         /* The rest of this function executes only for OST_READs */
1538
1539         /* if unwrap_bulk failed, return -EAGAIN to retry */
1540         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1541         if (rc < 0)
1542                 GOTO(out, rc = -EAGAIN);
1543
1544         if (rc > aa->aa_requested_nob) {
1545                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1546                        aa->aa_requested_nob);
1547                 RETURN(-EPROTO);
1548         }
1549
1550         if (rc != req->rq_bulk->bd_nob_transferred) {
1551                 CERROR ("Unexpected rc %d (%d transferred)\n",
1552                         rc, req->rq_bulk->bd_nob_transferred);
1553                 return (-EPROTO);
1554         }
1555
1556         if (rc < aa->aa_requested_nob)
1557                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1558
1559         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1560                 static int cksum_counter;
1561                 __u32      server_cksum = body->oa.o_cksum;
1562                 char      *via;
1563                 char      *router;
1564                 cksum_type_t cksum_type;
1565
1566                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
1567                         cksum_type = cksum_type_unpack(body->oa.o_flags);
1568                 else
1569                         cksum_type = OBD_CKSUM_CRC32;
1570                 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1571                                                  aa->aa_ppga, OST_READ,
1572                                                  cksum_type);
1573
1574                 if (peer->nid == req->rq_bulk->bd_sender) {
1575                         via = router = "";
1576                 } else {
1577                         via = " via ";
1578                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1579                 }
1580
1581                 if (server_cksum == ~0 && rc > 0) {
1582                         CERROR("Protocol error: server %s set the 'checksum' "
1583                                "bit, but didn't send a checksum.  Not fatal, "
1584                                "but please notify on http://bugzilla.lustre.org/\n",
1585                                libcfs_nid2str(peer->nid));
1586                 } else if (server_cksum != client_cksum) {
1587                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1588                                            "%s%s%s inode "DFID" object "
1589                                            LPU64"/"LPU64" extent "
1590                                            "["LPU64"-"LPU64"]\n",
1591                                            req->rq_import->imp_obd->obd_name,
1592                                            libcfs_nid2str(peer->nid),
1593                                            via, router,
1594                                            body->oa.o_valid & OBD_MD_FLFID ?
1595                                                 body->oa.o_parent_seq : (__u64)0,
1596                                            body->oa.o_valid & OBD_MD_FLFID ?
1597                                                 body->oa.o_parent_oid : 0,
1598                                            body->oa.o_valid & OBD_MD_FLFID ?
1599                                                 body->oa.o_parent_ver : 0,
1600                                            body->oa.o_id,
1601                                            body->oa.o_valid & OBD_MD_FLGROUP ?
1602                                                 body->oa.o_seq : (__u64)0,
1603                                            aa->aa_ppga[0]->off,
1604                                            aa->aa_ppga[aa->aa_page_count-1]->off +
1605                                            aa->aa_ppga[aa->aa_page_count-1]->count -
1606                                                                         1);
1607                         CERROR("client %x, server %x, cksum_type %x\n",
1608                                client_cksum, server_cksum, cksum_type);
1609                         cksum_counter = 0;
1610                         aa->aa_oa->o_cksum = client_cksum;
1611                         rc = -EAGAIN;
1612                 } else {
1613                         cksum_counter++;
1614                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1615                         rc = 0;
1616                 }
1617         } else if (unlikely(client_cksum)) {
1618                 static int cksum_missed;
1619
1620                 cksum_missed++;
1621                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1622                         CERROR("Checksum %u requested from %s but not sent\n",
1623                                cksum_missed, libcfs_nid2str(peer->nid));
1624         } else {
1625                 rc = 0;
1626         }
1627 out:
1628         if (rc >= 0)
1629                 lustre_get_wire_obdo(aa->aa_oa, &body->oa);
1630
1631         RETURN(rc);
1632 }
1633
1634 static int osc_brw_internal(int cmd, struct obd_export *exp, struct obdo *oa,
1635                             struct lov_stripe_md *lsm,
1636                             obd_count page_count, struct brw_page **pga,
1637                             struct obd_capa *ocapa)
1638 {
1639         struct ptlrpc_request *req;
1640         int                    rc;
1641         cfs_waitq_t            waitq;
1642         int                    resends = 0;
1643         struct l_wait_info     lwi;
1644
1645         ENTRY;
1646
1647         cfs_waitq_init(&waitq);
1648
1649 restart_bulk:
1650         rc = osc_brw_prep_request(cmd, &exp->exp_obd->u.cli, oa, lsm,
1651                                   page_count, pga, &req, ocapa, 0);
1652         if (rc != 0)
1653                 return (rc);
1654
1655         rc = ptlrpc_queue_wait(req);
1656
1657         if (rc == -ETIMEDOUT && req->rq_resend) {
1658                 DEBUG_REQ(D_HA, req,  "BULK TIMEOUT");
1659                 ptlrpc_req_finished(req);
1660                 goto restart_bulk;
1661         }
1662
1663         rc = osc_brw_fini_request(req, rc);
1664
1665         ptlrpc_req_finished(req);
1666         if (osc_recoverable_error(rc)) {
1667                 resends++;
1668                 if (!osc_should_resend(resends, &exp->exp_obd->u.cli)) {
1669                         CERROR("too many resend retries, returning error\n");
1670                         RETURN(-EIO);
1671                 }
1672
1673                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1674                 l_wait_event(waitq, 0, &lwi);
1675
1676                 goto restart_bulk;
1677         }
1678
1679         RETURN (rc);
1680 }
1681
1682 int osc_brw_redo_request(struct ptlrpc_request *request,
1683                          struct osc_brw_async_args *aa)
1684 {
1685         struct ptlrpc_request *new_req;
1686         struct ptlrpc_request_set *set = request->rq_set;
1687         struct osc_brw_async_args *new_aa;
1688         struct osc_async_page *oap;
1689         int rc = 0;
1690         ENTRY;
1691
1692         if (!osc_should_resend(aa->aa_resends, aa->aa_cli)) {
1693                 CERROR("too many resent retries, returning error\n");
1694                 RETURN(-EIO);
1695         }
1696
1697         DEBUG_REQ(D_ERROR, request, "redo for recoverable error");
1698
1699         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1700                                         OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1701                                   aa->aa_cli, aa->aa_oa,
1702                                   NULL /* lsm unused by osc currently */,
1703                                   aa->aa_page_count, aa->aa_ppga,
1704                                   &new_req, aa->aa_ocapa, 0);
1705         if (rc)
1706                 RETURN(rc);
1707
1708         client_obd_list_lock(&aa->aa_cli->cl_loi_list_lock);
1709
1710         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1711                 if (oap->oap_request != NULL) {
1712                         LASSERTF(request == oap->oap_request,
1713                                  "request %p != oap_request %p\n",
1714                                  request, oap->oap_request);
1715                         if (oap->oap_interrupted) {
1716                                 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1717                                 ptlrpc_req_finished(new_req);
1718                                 RETURN(-EINTR);
1719                         }
1720                 }
1721         }
1722         /* New request takes over pga and oaps from old request.
1723          * Note that copying a list_head doesn't work, need to move it... */
1724         aa->aa_resends++;
1725         new_req->rq_interpret_reply = request->rq_interpret_reply;
1726         new_req->rq_async_args = request->rq_async_args;
1727         new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1728
1729         new_aa = ptlrpc_req_async_args(new_req);
1730
1731         CFS_INIT_LIST_HEAD(&new_aa->aa_oaps);
1732         cfs_list_splice(&aa->aa_oaps, &new_aa->aa_oaps);
1733         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1734
1735         cfs_list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1736                 if (oap->oap_request) {
1737                         ptlrpc_req_finished(oap->oap_request);
1738                         oap->oap_request = ptlrpc_request_addref(new_req);
1739                 }
1740         }
1741
1742         new_aa->aa_ocapa = aa->aa_ocapa;
1743         aa->aa_ocapa = NULL;
1744
1745         /* use ptlrpc_set_add_req is safe because interpret functions work
1746          * in check_set context. only one way exist with access to request
1747          * from different thread got -EINTR - this way protected with
1748          * cl_loi_list_lock */
1749         ptlrpc_set_add_req(set, new_req);
1750
1751         client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1752
1753         DEBUG_REQ(D_INFO, new_req, "new request");
1754         RETURN(0);
1755 }
1756
1757 /*
1758  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1759  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1760  * fine for our small page arrays and doesn't require allocation.  its an
1761  * insertion sort that swaps elements that are strides apart, shrinking the
1762  * stride down until its '1' and the array is sorted.
1763  */
1764 static void sort_brw_pages(struct brw_page **array, int num)
1765 {
1766         int stride, i, j;
1767         struct brw_page *tmp;
1768
1769         if (num == 1)
1770                 return;
1771         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1772                 ;
1773
1774         do {
1775                 stride /= 3;
1776                 for (i = stride ; i < num ; i++) {
1777                         tmp = array[i];
1778                         j = i;
1779                         while (j >= stride && array[j - stride]->off > tmp->off) {
1780                                 array[j] = array[j - stride];
1781                                 j -= stride;
1782                         }
1783                         array[j] = tmp;
1784                 }
1785         } while (stride > 1);
1786 }
1787
1788 static obd_count max_unfragmented_pages(struct brw_page **pg, obd_count pages)
1789 {
1790         int count = 1;
1791         int offset;
1792         int i = 0;
1793
1794         LASSERT (pages > 0);
1795         offset = pg[i]->off & ~CFS_PAGE_MASK;
1796
1797         for (;;) {
1798                 pages--;
1799                 if (pages == 0)         /* that's all */
1800                         return count;
1801
1802                 if (offset + pg[i]->count < CFS_PAGE_SIZE)
1803                         return count;   /* doesn't end on page boundary */
1804
1805                 i++;
1806                 offset = pg[i]->off & ~CFS_PAGE_MASK;
1807                 if (offset != 0)        /* doesn't start on page boundary */
1808                         return count;
1809
1810                 count++;
1811         }
1812 }
1813
1814 static struct brw_page **osc_build_ppga(struct brw_page *pga, obd_count count)
1815 {
1816         struct brw_page **ppga;
1817         int i;
1818
1819         OBD_ALLOC(ppga, sizeof(*ppga) * count);
1820         if (ppga == NULL)
1821                 return NULL;
1822
1823         for (i = 0; i < count; i++)
1824                 ppga[i] = pga + i;
1825         return ppga;
1826 }
1827
1828 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1829 {
1830         LASSERT(ppga != NULL);
1831         OBD_FREE(ppga, sizeof(*ppga) * count);
1832 }
1833
1834 static int osc_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1835                    obd_count page_count, struct brw_page *pga,
1836                    struct obd_trans_info *oti)
1837 {
1838         struct obdo *saved_oa = NULL;
1839         struct brw_page **ppga, **orig;
1840         struct obd_import *imp = class_exp2cliimp(exp);
1841         struct client_obd *cli;
1842         int rc, page_count_orig;
1843         ENTRY;
1844
1845         LASSERT((imp != NULL) && (imp->imp_obd != NULL));
1846         cli = &imp->imp_obd->u.cli;
1847
1848         if (cmd & OBD_BRW_CHECK) {
1849                 /* The caller just wants to know if there's a chance that this
1850                  * I/O can succeed */
1851
1852                 if (imp->imp_invalid)
1853                         RETURN(-EIO);
1854                 RETURN(0);
1855         }
1856
1857         /* test_brw with a failed create can trip this, maybe others. */
1858         LASSERT(cli->cl_max_pages_per_rpc);
1859
1860         rc = 0;
1861
1862         orig = ppga = osc_build_ppga(pga, page_count);
1863         if (ppga == NULL)
1864                 RETURN(-ENOMEM);
1865         page_count_orig = page_count;
1866
1867         sort_brw_pages(ppga, page_count);
1868         while (page_count) {
1869                 obd_count pages_per_brw;
1870
1871                 if (page_count > cli->cl_max_pages_per_rpc)
1872                         pages_per_brw = cli->cl_max_pages_per_rpc;
1873                 else
1874                         pages_per_brw = page_count;
1875
1876                 pages_per_brw = max_unfragmented_pages(ppga, pages_per_brw);
1877
1878                 if (saved_oa != NULL) {
1879                         /* restore previously saved oa */
1880                         *oinfo->oi_oa = *saved_oa;
1881                 } else if (page_count > pages_per_brw) {
1882                         /* save a copy of oa (brw will clobber it) */
1883                         OBDO_ALLOC(saved_oa);
1884                         if (saved_oa == NULL)
1885                                 GOTO(out, rc = -ENOMEM);
1886                         *saved_oa = *oinfo->oi_oa;
1887                 }
1888
1889                 rc = osc_brw_internal(cmd, exp, oinfo->oi_oa, oinfo->oi_md,
1890                                       pages_per_brw, ppga, oinfo->oi_capa);
1891
1892                 if (rc != 0)
1893                         break;
1894
1895                 page_count -= pages_per_brw;
1896                 ppga += pages_per_brw;
1897         }
1898
1899 out:
1900         osc_release_ppga(orig, page_count_orig);
1901
1902         if (saved_oa != NULL)
1903                 OBDO_FREE(saved_oa);
1904
1905         RETURN(rc);
1906 }
1907
1908 /* The companion to osc_enter_cache(), called when @oap is no longer part of
1909  * the dirty accounting.  Writeback completes or truncate happens before
1910  * writing starts.  Must be called with the loi lock held. */
1911 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1912                            int sent)
1913 {
1914         osc_release_write_grant(cli, &oap->oap_brw_page, sent);
1915 }
1916
1917
1918 /* This maintains the lists of pending pages to read/write for a given object
1919  * (lop).  This is used by osc_check_rpcs->osc_next_loi() and loi_list_maint()
1920  * to quickly find objects that are ready to send an RPC. */
1921 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1922                          int cmd)
1923 {
1924         int optimal;
1925         ENTRY;
1926
1927         if (lop->lop_num_pending == 0)
1928                 RETURN(0);
1929
1930         /* if we have an invalid import we want to drain the queued pages
1931          * by forcing them through rpcs that immediately fail and complete
1932          * the pages.  recovery relies on this to empty the queued pages
1933          * before canceling the locks and evicting down the llite pages */
1934         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1935                 RETURN(1);
1936
1937         /* stream rpcs in queue order as long as as there is an urgent page
1938          * queued.  this is our cheap solution for good batching in the case
1939          * where writepage marks some random page in the middle of the file
1940          * as urgent because of, say, memory pressure */
1941         if (!cfs_list_empty(&lop->lop_urgent)) {
1942                 CDEBUG(D_CACHE, "urgent request forcing RPC\n");
1943                 RETURN(1);
1944         }
1945         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1946         optimal = cli->cl_max_pages_per_rpc;
1947         if (cmd & OBD_BRW_WRITE) {
1948                 /* trigger a write rpc stream as long as there are dirtiers
1949                  * waiting for space.  as they're waiting, they're not going to
1950                  * create more pages to coalesce with what's waiting.. */
1951                 if (!cfs_list_empty(&cli->cl_cache_waiters)) {
1952                         CDEBUG(D_CACHE, "cache waiters forcing RPC\n");
1953                         RETURN(1);
1954                 }
1955                 /* +16 to avoid triggering rpcs that would want to include pages
1956                  * that are being queued but which can't be made ready until
1957                  * the queuer finishes with the page. this is a wart for
1958                  * llite::commit_write() */
1959                 optimal += 16;
1960         }
1961         if (lop->lop_num_pending >= optimal)
1962                 RETURN(1);
1963
1964         RETURN(0);
1965 }
1966
1967 static int lop_makes_hprpc(struct loi_oap_pages *lop)
1968 {
1969         struct osc_async_page *oap;
1970         ENTRY;
1971
1972         if (cfs_list_empty(&lop->lop_urgent))
1973                 RETURN(0);
1974
1975         oap = cfs_list_entry(lop->lop_urgent.next,
1976                          struct osc_async_page, oap_urgent_item);
1977
1978         if (oap->oap_async_flags & ASYNC_HP) {
1979                 CDEBUG(D_CACHE, "hp request forcing RPC\n");
1980                 RETURN(1);
1981         }
1982
1983         RETURN(0);
1984 }
1985
1986 static void on_list(cfs_list_t *item, cfs_list_t *list,
1987                     int should_be_on)
1988 {
1989         if (cfs_list_empty(item) && should_be_on)
1990                 cfs_list_add_tail(item, list);
1991         else if (!cfs_list_empty(item) && !should_be_on)
1992                 cfs_list_del_init(item);
1993 }
1994
1995 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1996  * can find pages to build into rpcs quickly */
1997 void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1998 {
1999         if (lop_makes_hprpc(&loi->loi_write_lop) ||
2000             lop_makes_hprpc(&loi->loi_read_lop)) {
2001                 /* HP rpc */
2002                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list, 0);
2003                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 1);
2004         } else {
2005                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 0);
2006                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list,
2007                         lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)||
2008                         lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
2009         }
2010
2011         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
2012                 loi->loi_write_lop.lop_num_pending);
2013
2014         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
2015                 loi->loi_read_lop.lop_num_pending);
2016 }
2017
2018 static void lop_update_pending(struct client_obd *cli,
2019                                struct loi_oap_pages *lop, int cmd, int delta)
2020 {
2021         lop->lop_num_pending += delta;
2022         if (cmd & OBD_BRW_WRITE)
2023                 cli->cl_pending_w_pages += delta;
2024         else
2025                 cli->cl_pending_r_pages += delta;
2026 }
2027
2028 /**
2029  * this is called when a sync waiter receives an interruption.  Its job is to
2030  * get the caller woken as soon as possible.  If its page hasn't been put in an
2031  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
2032  * desiring interruption which will forcefully complete the rpc once the rpc
2033  * has timed out.
2034  */
2035 int osc_oap_interrupted(const struct lu_env *env, struct osc_async_page *oap)
2036 {
2037         struct loi_oap_pages *lop;
2038         struct lov_oinfo *loi;
2039         int rc = -EBUSY;
2040         ENTRY;
2041
2042         LASSERT(!oap->oap_interrupted);
2043         oap->oap_interrupted = 1;
2044
2045         /* ok, it's been put in an rpc. only one oap gets a request reference */
2046         if (oap->oap_request != NULL) {
2047                 ptlrpc_mark_interrupted(oap->oap_request);
2048                 ptlrpcd_wake(oap->oap_request);
2049                 ptlrpc_req_finished(oap->oap_request);
2050                 oap->oap_request = NULL;
2051         }
2052
2053         /*
2054          * page completion may be called only if ->cpo_prep() method was
2055          * executed by osc_io_submit(), that also adds page the to pending list
2056          */
2057         if (!cfs_list_empty(&oap->oap_pending_item)) {
2058                 cfs_list_del_init(&oap->oap_pending_item);
2059                 cfs_list_del_init(&oap->oap_urgent_item);
2060
2061                 loi = oap->oap_loi;
2062                 lop = (oap->oap_cmd & OBD_BRW_WRITE) ?
2063                         &loi->loi_write_lop : &loi->loi_read_lop;
2064                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
2065                 loi_list_maint(oap->oap_cli, oap->oap_loi);
2066                 rc = oap->oap_caller_ops->ap_completion(env,
2067                                           oap->oap_caller_data,
2068                                           oap->oap_cmd, NULL, -EINTR);
2069         }
2070
2071         RETURN(rc);
2072 }
2073
2074 /* this is trying to propogate async writeback errors back up to the
2075  * application.  As an async write fails we record the error code for later if
2076  * the app does an fsync.  As long as errors persist we force future rpcs to be
2077  * sync so that the app can get a sync error and break the cycle of queueing
2078  * pages for which writeback will fail. */
2079 static void osc_process_ar(struct osc_async_rc *ar, __u64 xid,
2080                            int rc)
2081 {
2082         if (rc) {
2083                 if (!ar->ar_rc)
2084                         ar->ar_rc = rc;
2085
2086                 ar->ar_force_sync = 1;
2087                 ar->ar_min_xid = ptlrpc_sample_next_xid();
2088                 return;
2089
2090         }
2091
2092         if (ar->ar_force_sync && (xid >= ar->ar_min_xid))
2093                 ar->ar_force_sync = 0;
2094 }
2095
2096 void osc_oap_to_pending(struct osc_async_page *oap)
2097 {
2098         struct loi_oap_pages *lop;
2099
2100         if (oap->oap_cmd & OBD_BRW_WRITE)
2101                 lop = &oap->oap_loi->loi_write_lop;
2102         else
2103                 lop = &oap->oap_loi->loi_read_lop;
2104
2105         if (oap->oap_async_flags & ASYNC_HP)
2106                 cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2107         else if (oap->oap_async_flags & ASYNC_URGENT)
2108                 cfs_list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2109         cfs_list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2110         lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, 1);
2111 }
2112
2113 /* this must be called holding the loi list lock to give coverage to exit_cache,
2114  * async_flag maintenance, and oap_request */
2115 static void osc_ap_completion(const struct lu_env *env,
2116                               struct client_obd *cli, struct obdo *oa,
2117                               struct osc_async_page *oap, int sent, int rc)
2118 {
2119         __u64 xid = 0;
2120
2121         ENTRY;
2122         if (oap->oap_request != NULL) {
2123                 xid = ptlrpc_req_xid(oap->oap_request);
2124                 ptlrpc_req_finished(oap->oap_request);
2125                 oap->oap_request = NULL;
2126         }
2127
2128         cfs_spin_lock(&oap->oap_lock);
2129         oap->oap_async_flags = 0;
2130         cfs_spin_unlock(&oap->oap_lock);
2131         oap->oap_interrupted = 0;
2132
2133         if (oap->oap_cmd & OBD_BRW_WRITE) {
2134                 osc_process_ar(&cli->cl_ar, xid, rc);
2135                 osc_process_ar(&oap->oap_loi->loi_ar, xid, rc);
2136         }
2137
2138         if (rc == 0 && oa != NULL) {
2139                 if (oa->o_valid & OBD_MD_FLBLOCKS)
2140                         oap->oap_loi->loi_lvb.lvb_blocks = oa->o_blocks;
2141                 if (oa->o_valid & OBD_MD_FLMTIME)
2142                         oap->oap_loi->loi_lvb.lvb_mtime = oa->o_mtime;
2143                 if (oa->o_valid & OBD_MD_FLATIME)
2144                         oap->oap_loi->loi_lvb.lvb_atime = oa->o_atime;
2145                 if (oa->o_valid & OBD_MD_FLCTIME)
2146                         oap->oap_loi->loi_lvb.lvb_ctime = oa->o_ctime;
2147         }
2148
2149         rc = oap->oap_caller_ops->ap_completion(env, oap->oap_caller_data,
2150                                                 oap->oap_cmd, oa, rc);
2151
2152         /* ll_ap_completion (from llite) drops PG_locked. so, a new
2153          * I/O on the page could start, but OSC calls it under lock
2154          * and thus we can add oap back to pending safely */
2155         if (rc)
2156                 /* upper layer wants to leave the page on pending queue */
2157                 osc_oap_to_pending(oap);
2158         else
2159                 osc_exit_cache(cli, oap, sent);
2160         EXIT;
2161 }
2162
2163 static int brw_interpret(const struct lu_env *env,
2164                          struct ptlrpc_request *req, void *data, int rc)
2165 {
2166         struct osc_brw_async_args *aa = data;
2167         struct client_obd *cli;
2168         int async;
2169         ENTRY;
2170
2171         rc = osc_brw_fini_request(req, rc);
2172         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2173         if (osc_recoverable_error(rc)) {
2174                 /* Only retry once for mmaped files since the mmaped page
2175                  * might be modified at anytime. We have to retry at least
2176                  * once in case there WAS really a corruption of the page
2177                  * on the network, that was not caused by mmap() modifying
2178                  * the page. Bug11742 */
2179                 if ((rc == -EAGAIN) && (aa->aa_resends > 0) &&
2180                     aa->aa_oa->o_valid & OBD_MD_FLFLAGS &&
2181                     aa->aa_oa->o_flags & OBD_FL_MMAP) {
2182                         rc = 0;
2183                 } else {
2184                         rc = osc_brw_redo_request(req, aa);
2185                         if (rc == 0)
2186                                 RETURN(0);
2187                 }
2188         }
2189
2190         if (aa->aa_ocapa) {
2191                 capa_put(aa->aa_ocapa);
2192                 aa->aa_ocapa = NULL;
2193         }
2194
2195         cli = aa->aa_cli;
2196
2197         client_obd_list_lock(&cli->cl_loi_list_lock);
2198
2199         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2200          * is called so we know whether to go to sync BRWs or wait for more
2201          * RPCs to complete */
2202         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2203                 cli->cl_w_in_flight--;
2204         else
2205                 cli->cl_r_in_flight--;
2206
2207         async = cfs_list_empty(&aa->aa_oaps);
2208         if (!async) { /* from osc_send_oap_rpc() */
2209                 struct osc_async_page *oap, *tmp;
2210                 /* the caller may re-use the oap after the completion call so
2211                  * we need to clean it up a little */
2212                 cfs_list_for_each_entry_safe(oap, tmp, &aa->aa_oaps,
2213                                              oap_rpc_item) {
2214                         cfs_list_del_init(&oap->oap_rpc_item);
2215                         osc_ap_completion(env, cli, aa->aa_oa, oap, 1, rc);
2216                 }
2217                 OBDO_FREE(aa->aa_oa);
2218         } else { /* from async_internal() */
2219                 obd_count i;
2220                 for (i = 0; i < aa->aa_page_count; i++)
2221                         osc_release_write_grant(aa->aa_cli, aa->aa_ppga[i], 1);
2222
2223                 if (aa->aa_oa->o_flags & OBD_FL_TEMPORARY)
2224                         OBDO_FREE(aa->aa_oa);
2225         }
2226         osc_wake_cache_waiters(cli);
2227         osc_check_rpcs(env, cli);
2228         client_obd_list_unlock(&cli->cl_loi_list_lock);
2229         if (!async)
2230                 cl_req_completion(env, aa->aa_clerq, rc);
2231         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2232
2233         RETURN(rc);
2234 }
2235
2236 static struct ptlrpc_request *osc_build_req(const struct lu_env *env,
2237                                             struct client_obd *cli,
2238                                             cfs_list_t *rpc_list,
2239                                             int page_count, int cmd)
2240 {
2241         struct ptlrpc_request *req;
2242         struct brw_page **pga = NULL;
2243         struct osc_brw_async_args *aa;
2244         struct obdo *oa = NULL;
2245         const struct obd_async_page_ops *ops = NULL;
2246         void *caller_data = NULL;
2247         struct osc_async_page *oap;
2248         struct osc_async_page *tmp;
2249         struct ost_body *body;
2250         struct cl_req *clerq = NULL;
2251         enum cl_req_type crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2252         struct ldlm_lock *lock = NULL;
2253         struct cl_req_attr crattr;
2254         int i, rc, mpflag = 0;
2255
2256         ENTRY;
2257         LASSERT(!cfs_list_empty(rpc_list));
2258
2259         if (cmd & OBD_BRW_MEMALLOC)
2260                 mpflag = cfs_memory_pressure_get_and_set();
2261
2262         memset(&crattr, 0, sizeof crattr);
2263         OBD_ALLOC(pga, sizeof(*pga) * page_count);
2264         if (pga == NULL)
2265                 GOTO(out, req = ERR_PTR(-ENOMEM));
2266
2267         OBDO_ALLOC(oa);
2268         if (oa == NULL)
2269                 GOTO(out, req = ERR_PTR(-ENOMEM));
2270
2271         i = 0;
2272         cfs_list_for_each_entry(oap, rpc_list, oap_rpc_item) {
2273                 struct cl_page *page = osc_oap2cl_page(oap);
2274                 if (ops == NULL) {
2275                         ops = oap->oap_caller_ops;
2276                         caller_data = oap->oap_caller_data;
2277
2278                         clerq = cl_req_alloc(env, page, crt,
2279                                              1 /* only 1-object rpcs for
2280                                                 * now */);
2281                         if (IS_ERR(clerq))
2282                                 GOTO(out, req = (void *)clerq);
2283                         lock = oap->oap_ldlm_lock;
2284                 }
2285                 pga[i] = &oap->oap_brw_page;
2286                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2287                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
2288                        pga[i]->pg, cfs_page_index(oap->oap_page), oap, pga[i]->flag);
2289                 i++;
2290                 cl_req_page_add(env, clerq, page);
2291         }
2292
2293         /* always get the data for the obdo for the rpc */
2294         LASSERT(ops != NULL);
2295         crattr.cra_oa = oa;
2296         crattr.cra_capa = NULL;
2297         cl_req_attr_set(env, clerq, &crattr, ~0ULL);
2298         if (lock) {
2299                 oa->o_handle = lock->l_remote_handle;
2300                 oa->o_valid |= OBD_MD_FLHANDLE;
2301         }
2302
2303         rc = cl_req_prep(env, clerq);
2304         if (rc != 0) {
2305                 CERROR("cl_req_prep failed: %d\n", rc);
2306                 GOTO(out, req = ERR_PTR(rc));
2307         }
2308
2309         sort_brw_pages(pga, page_count);
2310         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
2311                                   pga, &req, crattr.cra_capa, 1);
2312         if (rc != 0) {
2313                 CERROR("prep_req failed: %d\n", rc);
2314                 GOTO(out, req = ERR_PTR(rc));
2315         }
2316
2317         if (cmd & OBD_BRW_MEMALLOC)
2318                 req->rq_memalloc = 1;
2319
2320         /* Need to update the timestamps after the request is built in case
2321          * we race with setattr (locally or in queue at OST).  If OST gets
2322          * later setattr before earlier BRW (as determined by the request xid),
2323          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2324          * way to do this in a single call.  bug 10150 */
2325         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2326         cl_req_attr_set(env, clerq, &crattr,
2327                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2328
2329         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2330         aa = ptlrpc_req_async_args(req);
2331         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
2332         cfs_list_splice(rpc_list, &aa->aa_oaps);
2333         CFS_INIT_LIST_HEAD(rpc_list);
2334         aa->aa_clerq = clerq;
2335 out:
2336         if (cmd & OBD_BRW_MEMALLOC)
2337                 cfs_memory_pressure_restore(mpflag);
2338
2339         capa_put(crattr.cra_capa);
2340         if (IS_ERR(req)) {
2341                 if (oa)
2342                         OBDO_FREE(oa);
2343                 if (pga)
2344                         OBD_FREE(pga, sizeof(*pga) * page_count);
2345                 /* this should happen rarely and is pretty bad, it makes the
2346                  * pending list not follow the dirty order */
2347                 client_obd_list_lock(&cli->cl_loi_list_lock);
2348                 cfs_list_for_each_entry_safe(oap, tmp, rpc_list, oap_rpc_item) {
2349                         cfs_list_del_init(&oap->oap_rpc_item);
2350
2351                         /* queued sync pages can be torn down while the pages
2352                          * were between the pending list and the rpc */
2353                         if (oap->oap_interrupted) {
2354                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
2355                                 osc_ap_completion(env, cli, NULL, oap, 0,
2356                                                   oap->oap_count);
2357                                 continue;
2358                         }
2359                         osc_ap_completion(env, cli, NULL, oap, 0, PTR_ERR(req));
2360                 }
2361                 if (clerq && !IS_ERR(clerq))
2362                         cl_req_completion(env, clerq, PTR_ERR(req));
2363         }
2364         RETURN(req);
2365 }
2366
2367 /**
2368  * prepare pages for ASYNC io and put pages in send queue.
2369  *
2370  * \param cmd OBD_BRW_* macroses
2371  * \param lop pending pages
2372  *
2373  * \return zero if no page added to send queue.
2374  * \return 1 if pages successfully added to send queue.
2375  * \return negative on errors.
2376  */
2377 static int
2378 osc_send_oap_rpc(const struct lu_env *env, struct client_obd *cli,
2379                  struct lov_oinfo *loi,
2380                  int cmd, struct loi_oap_pages *lop)
2381 {
2382         struct ptlrpc_request *req;
2383         obd_count page_count = 0;
2384         struct osc_async_page *oap = NULL, *tmp;
2385         struct osc_brw_async_args *aa;
2386         const struct obd_async_page_ops *ops;
2387         CFS_LIST_HEAD(rpc_list);
2388         CFS_LIST_HEAD(tmp_list);
2389         unsigned int ending_offset;
2390         unsigned  starting_offset = 0;
2391         int srvlock = 0, mem_tight = 0;
2392         struct cl_object *clob = NULL;
2393         ENTRY;
2394
2395         /* ASYNC_HP pages first. At present, when the lock the pages is
2396          * to be canceled, the pages covered by the lock will be sent out
2397          * with ASYNC_HP. We have to send out them as soon as possible. */
2398         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_urgent, oap_urgent_item) {
2399                 if (oap->oap_async_flags & ASYNC_HP) 
2400                         cfs_list_move(&oap->oap_pending_item, &tmp_list);
2401                 else
2402                         cfs_list_move_tail(&oap->oap_pending_item, &tmp_list);
2403                 if (++page_count >= cli->cl_max_pages_per_rpc)
2404                         break;
2405         }
2406
2407         cfs_list_splice(&tmp_list, &lop->lop_pending);
2408         page_count = 0;
2409
2410         /* first we find the pages we're allowed to work with */
2411         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_pending,
2412                                      oap_pending_item) {
2413                 ops = oap->oap_caller_ops;
2414
2415                 LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
2416                          "magic 0x%x\n", oap, oap->oap_magic);
2417
2418                 if (clob == NULL) {
2419                         /* pin object in memory, so that completion call-backs
2420                          * can be safely called under client_obd_list lock. */
2421                         clob = osc_oap2cl_page(oap)->cp_obj;
2422                         cl_object_get(clob);
2423                 }
2424
2425                 if (page_count != 0 &&
2426                     srvlock != !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK)) {
2427                         CDEBUG(D_PAGE, "SRVLOCK flag mismatch,"
2428                                " oap %p, page %p, srvlock %u\n",
2429                                oap, oap->oap_brw_page.pg, (unsigned)!srvlock);
2430                         break;
2431                 }
2432
2433                 /* If there is a gap at the start of this page, it can't merge
2434                  * with any previous page, so we'll hand the network a
2435                  * "fragmented" page array that it can't transfer in 1 RDMA */
2436                 if (page_count != 0 && oap->oap_page_off != 0)
2437                         break;
2438
2439                 /* in llite being 'ready' equates to the page being locked
2440                  * until completion unlocks it.  commit_write submits a page
2441                  * as not ready because its unlock will happen unconditionally
2442                  * as the call returns.  if we race with commit_write giving
2443                  * us that page we don't want to create a hole in the page
2444                  * stream, so we stop and leave the rpc to be fired by
2445                  * another dirtier or kupdated interval (the not ready page
2446                  * will still be on the dirty list).  we could call in
2447                  * at the end of ll_file_write to process the queue again. */
2448                 if (!(oap->oap_async_flags & ASYNC_READY)) {
2449                         int rc = ops->ap_make_ready(env, oap->oap_caller_data,
2450                                                     cmd);
2451                         if (rc < 0)
2452                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
2453                                                 "instead of ready\n", oap,
2454                                                 oap->oap_page, rc);
2455                         switch (rc) {
2456                         case -EAGAIN:
2457                                 /* llite is telling us that the page is still
2458                                  * in commit_write and that we should try
2459                                  * and put it in an rpc again later.  we
2460                                  * break out of the loop so we don't create
2461                                  * a hole in the sequence of pages in the rpc
2462                                  * stream.*/
2463                                 oap = NULL;
2464                                 break;
2465                         case -EINTR:
2466                                 /* the io isn't needed.. tell the checks
2467                                  * below to complete the rpc with EINTR */
2468                                 cfs_spin_lock(&oap->oap_lock);
2469                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
2470                                 cfs_spin_unlock(&oap->oap_lock);
2471                                 oap->oap_count = -EINTR;
2472                                 break;
2473                         case 0:
2474                                 cfs_spin_lock(&oap->oap_lock);
2475                                 oap->oap_async_flags |= ASYNC_READY;
2476                                 cfs_spin_unlock(&oap->oap_lock);
2477                                 break;
2478                         default:
2479                                 LASSERTF(0, "oap %p page %p returned %d "
2480                                             "from make_ready\n", oap,
2481                                             oap->oap_page, rc);
2482                                 break;
2483                         }
2484                 }
2485                 if (oap == NULL)
2486                         break;
2487                 /*
2488                  * Page submitted for IO has to be locked. Either by
2489                  * ->ap_make_ready() or by higher layers.
2490                  */
2491 #if defined(__KERNEL__) && defined(__linux__)
2492                 {
2493                         struct cl_page *page;
2494
2495                         page = osc_oap2cl_page(oap);
2496
2497                         if (page->cp_type == CPT_CACHEABLE &&
2498                             !(PageLocked(oap->oap_page) &&
2499                               (CheckWriteback(oap->oap_page, cmd)))) {
2500                                 CDEBUG(D_PAGE, "page %p lost wb %lx/%x\n",
2501                                        oap->oap_page,
2502                                        (long)oap->oap_page->flags,
2503                                        oap->oap_async_flags);
2504                                 LBUG();
2505                         }
2506                 }
2507 #endif
2508
2509                 /* take the page out of our book-keeping */
2510                 cfs_list_del_init(&oap->oap_pending_item);
2511                 lop_update_pending(cli, lop, cmd, -1);
2512                 cfs_list_del_init(&oap->oap_urgent_item);
2513
2514                 if (page_count == 0)
2515                         starting_offset = (oap->oap_obj_off+oap->oap_page_off) &
2516                                           (PTLRPC_MAX_BRW_SIZE - 1);
2517
2518                 /* ask the caller for the size of the io as the rpc leaves. */
2519                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE)) {
2520                         oap->oap_count =
2521                                 ops->ap_refresh_count(env, oap->oap_caller_data,
2522                                                       cmd);
2523                         LASSERT(oap->oap_page_off + oap->oap_count <= CFS_PAGE_SIZE);
2524                 }
2525                 if (oap->oap_count <= 0) {
2526                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
2527                                oap->oap_count);
2528                         osc_ap_completion(env, cli, NULL,
2529                                           oap, 0, oap->oap_count);
2530                         continue;
2531                 }
2532
2533                 /* now put the page back in our accounting */
2534                 cfs_list_add_tail(&oap->oap_rpc_item, &rpc_list);
2535                 if (oap->oap_brw_flags & OBD_BRW_MEMALLOC)
2536                         mem_tight = 1;
2537                 if (page_count == 0)
2538                         srvlock = !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK);
2539                 if (++page_count >= cli->cl_max_pages_per_rpc)
2540                         break;
2541
2542                 /* End on a PTLRPC_MAX_BRW_SIZE boundary.  We want full-sized
2543                  * RPCs aligned on PTLRPC_MAX_BRW_SIZE boundaries to help reads
2544                  * have the same alignment as the initial writes that allocated
2545                  * extents on the server. */
2546                 ending_offset = (oap->oap_obj_off + oap->oap_page_off +
2547                                  oap->oap_count) & (PTLRPC_MAX_BRW_SIZE - 1);
2548                 if (ending_offset == 0)
2549                         break;
2550
2551                 /* If there is a gap at the end of this page, it can't merge
2552                  * with any subsequent pages, so we'll hand the network a
2553                  * "fragmented" page array that it can't transfer in 1 RDMA */
2554                 if (oap->oap_page_off + oap->oap_count < CFS_PAGE_SIZE)
2555                         break;
2556         }
2557
2558         osc_wake_cache_waiters(cli);
2559
2560         loi_list_maint(cli, loi);
2561
2562         client_obd_list_unlock(&cli->cl_loi_list_lock);
2563
2564         if (clob != NULL)
2565                 cl_object_put(env, clob);
2566
2567         if (page_count == 0) {
2568                 client_obd_list_lock(&cli->cl_loi_list_lock);
2569                 RETURN(0);
2570         }
2571
2572         req = osc_build_req(env, cli, &rpc_list, page_count,
2573                             mem_tight ? (cmd | OBD_BRW_MEMALLOC) : cmd);
2574         if (IS_ERR(req)) {
2575                 LASSERT(cfs_list_empty(&rpc_list));
2576                 loi_list_maint(cli, loi);
2577                 RETURN(PTR_ERR(req));
2578         }
2579
2580         aa = ptlrpc_req_async_args(req);
2581
2582         if (cmd == OBD_BRW_READ) {
2583                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2584                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2585                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2586                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2587         } else {
2588                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2589                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
2590                                  cli->cl_w_in_flight);
2591                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2592                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2593         }
2594         ptlrpc_lprocfs_brw(req, aa->aa_requested_nob);
2595
2596         client_obd_list_lock(&cli->cl_loi_list_lock);
2597
2598         if (cmd == OBD_BRW_READ)
2599                 cli->cl_r_in_flight++;
2600         else
2601                 cli->cl_w_in_flight++;
2602
2603         /* queued sync pages can be torn down while the pages
2604          * were between the pending list and the rpc */
2605         tmp = NULL;
2606         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2607                 /* only one oap gets a request reference */
2608                 if (tmp == NULL)
2609                         tmp = oap;
2610                 if (oap->oap_interrupted && !req->rq_intr) {
2611                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2612                                oap, req);
2613                         ptlrpc_mark_interrupted(req);
2614                 }
2615         }
2616         if (tmp != NULL)
2617                 tmp->oap_request = ptlrpc_request_addref(req);
2618
2619         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2620                   page_count, aa, cli->cl_r_in_flight, cli->cl_w_in_flight);
2621
2622         req->rq_interpret_reply = brw_interpret;
2623         ptlrpcd_add_req(req, PSCOPE_BRW);
2624         RETURN(1);
2625 }
2626
2627 #define LOI_DEBUG(LOI, STR, args...)                                     \
2628         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
2629                !cfs_list_empty(&(LOI)->loi_ready_item) ||                \
2630                !cfs_list_empty(&(LOI)->loi_hp_ready_item),               \
2631                (LOI)->loi_write_lop.lop_num_pending,                     \
2632                !cfs_list_empty(&(LOI)->loi_write_lop.lop_urgent),        \
2633                (LOI)->loi_read_lop.lop_num_pending,                      \
2634                !cfs_list_empty(&(LOI)->loi_read_lop.lop_urgent),         \
2635                args)                                                     \
2636
2637 /* This is called by osc_check_rpcs() to find which objects have pages that
2638  * we could be sending.  These lists are maintained by lop_makes_rpc(). */
2639 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
2640 {
2641         ENTRY;
2642
2643         /* First return objects that have blocked locks so that they
2644          * will be flushed quickly and other clients can get the lock,
2645          * then objects which have pages ready to be stuffed into RPCs */
2646         if (!cfs_list_empty(&cli->cl_loi_hp_ready_list))
2647                 RETURN(cfs_list_entry(cli->cl_loi_hp_ready_list.next,
2648                                       struct lov_oinfo, loi_hp_ready_item));
2649         if (!cfs_list_empty(&cli->cl_loi_ready_list))
2650                 RETURN(cfs_list_entry(cli->cl_loi_ready_list.next,
2651                                       struct lov_oinfo, loi_ready_item));
2652
2653         /* then if we have cache waiters, return all objects with queued
2654          * writes.  This is especially important when many small files
2655          * have filled up the cache and not been fired into rpcs because
2656          * they don't pass the nr_pending/object threshhold */
2657         if (!cfs_list_empty(&cli->cl_cache_waiters) &&
2658             !cfs_list_empty(&cli->cl_loi_write_list))
2659                 RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2660                                       struct lov_oinfo, loi_write_item));
2661
2662         /* then return all queued objects when we have an invalid import
2663          * so that they get flushed */
2664         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
2665                 if (!cfs_list_empty(&cli->cl_loi_write_list))
2666                         RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2667                                               struct lov_oinfo,
2668                                               loi_write_item));
2669                 if (!cfs_list_empty(&cli->cl_loi_read_list))
2670                         RETURN(cfs_list_entry(cli->cl_loi_read_list.next,
2671                                               struct lov_oinfo, loi_read_item));
2672         }
2673         RETURN(NULL);
2674 }
2675
2676 static int osc_max_rpc_in_flight(struct client_obd *cli, struct lov_oinfo *loi)
2677 {
2678         struct osc_async_page *oap;
2679         int hprpc = 0;
2680
2681         if (!cfs_list_empty(&loi->loi_write_lop.lop_urgent)) {
2682                 oap = cfs_list_entry(loi->loi_write_lop.lop_urgent.next,
2683                                      struct osc_async_page, oap_urgent_item);
2684                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2685         }
2686
2687         if (!hprpc && !cfs_list_empty(&loi->loi_read_lop.lop_urgent)) {
2688                 oap = cfs_list_entry(loi->loi_read_lop.lop_urgent.next,
2689                                      struct osc_async_page, oap_urgent_item);
2690                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2691         }
2692
2693         return rpcs_in_flight(cli) >= cli->cl_max_rpcs_in_flight + hprpc;
2694 }
2695
2696 /* called with the loi list lock held */
2697 void osc_check_rpcs(const struct lu_env *env, struct client_obd *cli)
2698 {
2699         struct lov_oinfo *loi;
2700         int rc = 0, race_counter = 0;
2701         ENTRY;
2702
2703         while ((loi = osc_next_loi(cli)) != NULL) {
2704                 LOI_DEBUG(loi, "%lu in flight\n", rpcs_in_flight(cli));
2705
2706                 if (osc_max_rpc_in_flight(cli, loi))
2707                         break;
2708
2709                 /* attempt some read/write balancing by alternating between
2710                  * reads and writes in an object.  The makes_rpc checks here
2711                  * would be redundant if we were getting read/write work items
2712                  * instead of objects.  we don't want send_oap_rpc to drain a
2713                  * partial read pending queue when we're given this object to
2714                  * do io on writes while there are cache waiters */
2715                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
2716                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_WRITE,
2717                                               &loi->loi_write_lop);
2718                         if (rc < 0) {
2719                                 CERROR("Write request failed with %d\n", rc);
2720
2721                                 /* osc_send_oap_rpc failed, mostly because of
2722                                  * memory pressure.
2723                                  *
2724                                  * It can't break here, because if:
2725                                  *  - a page was submitted by osc_io_submit, so
2726                                  *    page locked;
2727                                  *  - no request in flight
2728                                  *  - no subsequent request
2729                                  * The system will be in live-lock state,
2730                                  * because there is no chance to call
2731                                  * osc_io_unplug() and osc_check_rpcs() any
2732                                  * more. pdflush can't help in this case,
2733                                  * because it might be blocked at grabbing
2734                                  * the page lock as we mentioned.
2735                                  *
2736                                  * Anyway, continue to drain pages. */
2737                                 /* break; */
2738                         }
2739
2740                         if (rc > 0)
2741                                 race_counter = 0;
2742                         else
2743                                 race_counter++;
2744                 }
2745                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
2746                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_READ,
2747                                               &loi->loi_read_lop);
2748                         if (rc < 0)
2749                                 CERROR("Read request failed with %d\n", rc);
2750
2751                         if (rc > 0)
2752                                 race_counter = 0;
2753                         else
2754                                 race_counter++;
2755                 }
2756
2757                 /* attempt some inter-object balancing by issuing rpcs
2758                  * for each object in turn */
2759                 if (!cfs_list_empty(&loi->loi_hp_ready_item))
2760                         cfs_list_del_init(&loi->loi_hp_ready_item);
2761                 if (!cfs_list_empty(&loi->loi_ready_item))
2762                         cfs_list_del_init(&loi->loi_ready_item);
2763                 if (!cfs_list_empty(&loi->loi_write_item))
2764                         cfs_list_del_init(&loi->loi_write_item);
2765                 if (!cfs_list_empty(&loi->loi_read_item))
2766                         cfs_list_del_init(&loi->loi_read_item);
2767
2768                 loi_list_maint(cli, loi);
2769
2770                 /* send_oap_rpc fails with 0 when make_ready tells it to
2771                  * back off.  llite's make_ready does this when it tries
2772                  * to lock a page queued for write that is already locked.
2773                  * we want to try sending rpcs from many objects, but we
2774                  * don't want to spin failing with 0.  */
2775                 if (race_counter == 10)
2776                         break;
2777         }
2778         EXIT;
2779 }
2780
2781 /* we're trying to queue a page in the osc so we're subject to the
2782  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
2783  * If the osc's queued pages are already at that limit, then we want to sleep
2784  * until there is space in the osc's queue for us.  We also may be waiting for
2785  * write credits from the OST if there are RPCs in flight that may return some
2786  * before we fall back to sync writes.
2787  *
2788  * We need this know our allocation was granted in the presence of signals */
2789 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
2790 {
2791         int rc;
2792         ENTRY;
2793         client_obd_list_lock(&cli->cl_loi_list_lock);
2794         rc = cfs_list_empty(&ocw->ocw_entry) || rpcs_in_flight(cli) == 0;
2795         client_obd_list_unlock(&cli->cl_loi_list_lock);
2796         RETURN(rc);
2797 };
2798
2799 /**
2800  * Non-blocking version of osc_enter_cache() that consumes grant only when it
2801  * is available.
2802  */
2803 int osc_enter_cache_try(const struct lu_env *env,
2804                         struct client_obd *cli, struct lov_oinfo *loi,
2805                         struct osc_async_page *oap, int transient)
2806 {
2807         int has_grant;
2808
2809         has_grant = cli->cl_avail_grant >= CFS_PAGE_SIZE;
2810         if (has_grant) {
2811                 osc_consume_write_grant(cli, &oap->oap_brw_page);
2812                 if (transient) {
2813                         cli->cl_dirty_transit += CFS_PAGE_SIZE;
2814                         cfs_atomic_inc(&obd_dirty_transit_pages);
2815                         oap->oap_brw_flags |= OBD_BRW_NOCACHE;
2816                 }
2817         }
2818         return has_grant;
2819 }
2820
2821 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
2822  * grant or cache space. */
2823 static int osc_enter_cache(const struct lu_env *env,
2824                            struct client_obd *cli, struct lov_oinfo *loi,
2825                            struct osc_async_page *oap)
2826 {
2827         struct osc_cache_waiter ocw;
2828         struct l_wait_info lwi = { 0 };
2829
2830         ENTRY;
2831
2832         CDEBUG(D_CACHE, "dirty: %ld/%d dirty_max: %ld/%d dropped: %lu "
2833                "grant: %lu\n", cli->cl_dirty, cfs_atomic_read(&obd_dirty_pages),
2834                cli->cl_dirty_max, obd_max_dirty_pages,
2835                cli->cl_lost_grant, cli->cl_avail_grant);
2836
2837         /* force the caller to try sync io.  this can jump the list
2838          * of queued writes and create a discontiguous rpc stream */
2839         if (cli->cl_dirty_max < CFS_PAGE_SIZE || cli->cl_ar.ar_force_sync ||
2840             loi->loi_ar.ar_force_sync)
2841                 RETURN(-EDQUOT);
2842
2843         /* Hopefully normal case - cache space and write credits available */
2844         if (cli->cl_dirty + CFS_PAGE_SIZE <= cli->cl_dirty_max &&
2845             cfs_atomic_read(&obd_dirty_pages) + 1 <= obd_max_dirty_pages &&
2846             osc_enter_cache_try(env, cli, loi, oap, 0))
2847                 RETURN(0);
2848
2849         /* It is safe to block as a cache waiter as long as there is grant
2850          * space available or the hope of additional grant being returned
2851          * when an in flight write completes.  Using the write back cache
2852          * if possible is preferable to sending the data synchronously
2853          * because write pages can then be merged in to large requests.
2854          * The addition of this cache waiter will causing pending write
2855          * pages to be sent immediately. */
2856         if (cli->cl_w_in_flight || cli->cl_avail_grant >= CFS_PAGE_SIZE) {
2857                 cfs_list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
2858                 cfs_waitq_init(&ocw.ocw_waitq);
2859                 ocw.ocw_oap = oap;
2860                 ocw.ocw_rc = 0;
2861
2862                 loi_list_maint(cli, loi);
2863                 osc_check_rpcs(env, cli);
2864                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2865
2866                 CDEBUG(D_CACHE, "sleeping for cache space\n");
2867                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
2868
2869                 client_obd_list_lock(&cli->cl_loi_list_lock);
2870                 if (!cfs_list_empty(&ocw.ocw_entry)) {
2871                         cfs_list_del(&ocw.ocw_entry);
2872                         RETURN(-EINTR);
2873                 }
2874                 RETURN(ocw.ocw_rc);
2875         }
2876
2877         RETURN(-EDQUOT);
2878 }
2879
2880
2881 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
2882                         struct lov_oinfo *loi, cfs_page_t *page,
2883                         obd_off offset, const struct obd_async_page_ops *ops,
2884                         void *data, void **res, int nocache,
2885                         struct lustre_handle *lockh)
2886 {
2887         struct osc_async_page *oap;
2888
2889         ENTRY;
2890
2891         if (!page)
2892                 return cfs_size_round(sizeof(*oap));
2893
2894         oap = *res;
2895         oap->oap_magic = OAP_MAGIC;
2896         oap->oap_cli = &exp->exp_obd->u.cli;
2897         oap->oap_loi = loi;
2898
2899         oap->oap_caller_ops = ops;
2900         oap->oap_caller_data = data;
2901
2902         oap->oap_page = page;
2903         oap->oap_obj_off = offset;
2904         if (!client_is_remote(exp) &&
2905             cfs_capable(CFS_CAP_SYS_RESOURCE))
2906                 oap->oap_brw_flags = OBD_BRW_NOQUOTA;
2907
2908         LASSERT(!(offset & ~CFS_PAGE_MASK));
2909
2910         CFS_INIT_LIST_HEAD(&oap->oap_pending_item);
2911         CFS_INIT_LIST_HEAD(&oap->oap_urgent_item);
2912         CFS_INIT_LIST_HEAD(&oap->oap_rpc_item);
2913         CFS_INIT_LIST_HEAD(&oap->oap_page_list);
2914
2915         cfs_spin_lock_init(&oap->oap_lock);
2916         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
2917         RETURN(0);
2918 }
2919
2920 struct osc_async_page *oap_from_cookie(void *cookie)
2921 {
2922         struct osc_async_page *oap = cookie;
2923         if (oap->oap_magic != OAP_MAGIC)
2924                 return ERR_PTR(-EINVAL);
2925         return oap;
2926 };
2927
2928 int osc_queue_async_io(const struct lu_env *env,
2929                        struct obd_export *exp, struct lov_stripe_md *lsm,
2930                        struct lov_oinfo *loi, void *cookie,
2931                        int cmd, obd_off off, int count,
2932                        obd_flag brw_flags, enum async_flags async_flags)
2933 {
2934         struct client_obd *cli = &exp->exp_obd->u.cli;
2935         struct osc_async_page *oap;
2936         int rc = 0;
2937         ENTRY;
2938
2939         oap = oap_from_cookie(cookie);
2940         if (IS_ERR(oap))
2941                 RETURN(PTR_ERR(oap));
2942
2943         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2944                 RETURN(-EIO);
2945
2946         if (!cfs_list_empty(&oap->oap_pending_item) ||
2947             !cfs_list_empty(&oap->oap_urgent_item) ||
2948             !cfs_list_empty(&oap->oap_rpc_item))
2949                 RETURN(-EBUSY);
2950
2951         /* check if the file's owner/group is over quota */
2952         if ((cmd & OBD_BRW_WRITE) && !(cmd & OBD_BRW_NOQUOTA)) {
2953                 struct cl_object *obj;
2954                 struct cl_attr    attr; /* XXX put attr into thread info */
2955                 unsigned int qid[MAXQUOTAS];
2956
2957                 obj = cl_object_top(osc_oap2cl_page(oap)->cp_obj);
2958
2959                 cl_object_attr_lock(obj);
2960                 rc = cl_object_attr_get(env, obj, &attr);
2961                 cl_object_attr_unlock(obj);
2962
2963                 qid[USRQUOTA] = attr.cat_uid;
2964                 qid[GRPQUOTA] = attr.cat_gid;
2965                 if (rc == 0 &&
2966                     lquota_chkdq(quota_interface, cli, qid) == NO_QUOTA)
2967                         rc = -EDQUOT;
2968                 if (rc)
2969                         RETURN(rc);
2970         }
2971
2972         if (loi == NULL)
2973                 loi = lsm->lsm_oinfo[0];
2974
2975         client_obd_list_lock(&cli->cl_loi_list_lock);
2976
2977         LASSERT(off + count <= CFS_PAGE_SIZE);
2978         oap->oap_cmd = cmd;
2979         oap->oap_page_off = off;
2980         oap->oap_count = count;
2981         oap->oap_brw_flags = brw_flags;
2982         /* Give a hint to OST that requests are coming from kswapd - bug19529 */
2983         if (cfs_memory_pressure_get())
2984                 oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
2985         cfs_spin_lock(&oap->oap_lock);
2986         oap->oap_async_flags = async_flags;
2987         cfs_spin_unlock(&oap->oap_lock);
2988
2989         if (cmd & OBD_BRW_WRITE) {
2990                 rc = osc_enter_cache(env, cli, loi, oap);
2991                 if (rc) {
2992                         client_obd_list_unlock(&cli->cl_loi_list_lock);
2993                         RETURN(rc);
2994                 }
2995         }
2996
2997         osc_oap_to_pending(oap);
2998         loi_list_maint(cli, loi);
2999
3000         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
3001                   cmd);
3002
3003         osc_check_rpcs(env, cli);
3004         client_obd_list_unlock(&cli->cl_loi_list_lock);
3005
3006         RETURN(0);
3007 }
3008
3009 /* aka (~was & now & flag), but this is more clear :) */
3010 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
3011
3012 int osc_set_async_flags_base(struct client_obd *cli,
3013                              struct lov_oinfo *loi, struct osc_async_page *oap,
3014                              obd_flag async_flags)
3015 {
3016         struct loi_oap_pages *lop;
3017         int flags = 0;
3018         ENTRY;
3019
3020         LASSERT(!cfs_list_empty(&oap->oap_pending_item));
3021
3022         if (oap->oap_cmd & OBD_BRW_WRITE) {
3023                 lop = &loi->loi_write_lop;
3024         } else {
3025                 lop = &loi->loi_read_lop;
3026         }
3027
3028         if ((oap->oap_async_flags & async_flags) == async_flags)
3029                 RETURN(0);
3030
3031         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
3032                 flags |= ASYNC_READY;
3033
3034         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT) &&
3035             cfs_list_empty(&oap->oap_rpc_item)) {
3036                 if (oap->oap_async_flags & ASYNC_HP)
3037                         cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
3038                 else
3039                         cfs_list_add_tail(&oap->oap_urgent_item,
3040                                           &lop->lop_urgent);
3041                 flags |= ASYNC_URGENT;
3042                 loi_list_maint(cli, loi);
3043         }
3044         cfs_spin_lock(&oap->oap_lock);
3045         oap->oap_async_flags |= flags;
3046         cfs_spin_unlock(&oap->oap_lock);
3047
3048         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
3049                         oap->oap_async_flags);
3050         RETURN(0);
3051 }
3052
3053 int osc_teardown_async_page(struct obd_export *exp,
3054                             struct lov_stripe_md *lsm,
3055                             struct lov_oinfo *loi, void *cookie)
3056 {
3057         struct client_obd *cli = &exp->exp_obd->u.cli;
3058         struct loi_oap_pages *lop;
3059         struct osc_async_page *oap;
3060         int rc = 0;
3061         ENTRY;
3062
3063         oap = oap_from_cookie(cookie);
3064         if (IS_ERR(oap))
3065                 RETURN(PTR_ERR(oap));
3066
3067         if (loi == NULL)
3068                 loi = lsm->lsm_oinfo[0];
3069
3070         if (oap->oap_cmd & OBD_BRW_WRITE) {
3071                 lop = &loi->loi_write_lop;
3072         } else {
3073                 lop = &loi->loi_read_lop;
3074         }
3075
3076         client_obd_list_lock(&cli->cl_loi_list_lock);
3077
3078         if (!cfs_list_empty(&oap->oap_rpc_item))
3079                 GOTO(out, rc = -EBUSY);
3080
3081         osc_exit_cache(cli, oap, 0);
3082         osc_wake_cache_waiters(cli);
3083
3084         if (!cfs_list_empty(&oap->oap_urgent_item)) {
3085                 cfs_list_del_init(&oap->oap_urgent_item);
3086                 cfs_spin_lock(&oap->oap_lock);
3087                 oap->oap_async_flags &= ~(ASYNC_URGENT | ASYNC_HP);
3088                 cfs_spin_unlock(&oap->oap_lock);
3089         }
3090         if (!cfs_list_empty(&oap->oap_pending_item)) {
3091                 cfs_list_del_init(&oap->oap_pending_item);
3092                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
3093         }
3094         loi_list_maint(cli, loi);
3095         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
3096 out:
3097         client_obd_list_unlock(&cli->cl_loi_list_lock);
3098         RETURN(rc);
3099 }
3100
3101 static void osc_set_lock_data_with_check(struct ldlm_lock *lock,
3102                                          struct ldlm_enqueue_info *einfo,
3103                                          int flags)
3104 {
3105         void *data = einfo->ei_cbdata;
3106
3107         LASSERT(lock != NULL);
3108         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
3109         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
3110         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
3111         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
3112
3113         lock_res_and_lock(lock);
3114         cfs_spin_lock(&osc_ast_guard);
3115         LASSERT(lock->l_ast_data == NULL || lock->l_ast_data == data);
3116         lock->l_ast_data = data;
3117         cfs_spin_unlock(&osc_ast_guard);
3118         unlock_res_and_lock(lock);
3119 }
3120
3121 static void osc_set_data_with_check(struct lustre_handle *lockh,
3122                                     struct ldlm_enqueue_info *einfo,
3123                                     int flags)
3124 {
3125         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
3126
3127         if (lock != NULL) {
3128                 osc_set_lock_data_with_check(lock, einfo, flags);
3129                 LDLM_LOCK_PUT(lock);
3130         } else
3131                 CERROR("lockh %p, data %p - client evicted?\n",
3132                        lockh, einfo->ei_cbdata);
3133 }
3134
3135 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3136                              ldlm_iterator_t replace, void *data)
3137 {
3138         struct ldlm_res_id res_id;
3139         struct obd_device *obd = class_exp2obd(exp);
3140
3141         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3142         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3143         return 0;
3144 }
3145
3146 /* find any ldlm lock of the inode in osc
3147  * return 0    not find
3148  *        1    find one
3149  *      < 0    error */
3150 static int osc_find_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3151                            ldlm_iterator_t replace, void *data)
3152 {
3153         struct ldlm_res_id res_id;
3154         struct obd_device *obd = class_exp2obd(exp);
3155         int rc = 0;
3156
3157         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3158         rc = ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3159         if (rc == LDLM_ITER_STOP)
3160                 return(1);
3161         if (rc == LDLM_ITER_CONTINUE)
3162                 return(0);
3163         return(rc);
3164 }
3165
3166 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
3167                             obd_enqueue_update_f upcall, void *cookie,
3168                             int *flags, int rc)
3169 {
3170         int intent = *flags & LDLM_FL_HAS_INTENT;
3171         ENTRY;
3172
3173         if (intent) {
3174                 /* The request was created before ldlm_cli_enqueue call. */
3175                 if (rc == ELDLM_LOCK_ABORTED) {
3176                         struct ldlm_reply *rep;
3177                         rep = req_capsule_server_get(&req->rq_pill,
3178                                                      &RMF_DLM_REP);
3179
3180                         LASSERT(rep != NULL);
3181                         if (rep->lock_policy_res1)
3182                                 rc = rep->lock_policy_res1;
3183                 }
3184         }
3185
3186         if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3187                 *flags |= LDLM_FL_LVB_READY;
3188                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3189                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
3190         }
3191
3192         /* Call the update callback. */
3193         rc = (*upcall)(cookie, rc);
3194         RETURN(rc);
3195 }
3196
3197 static int osc_enqueue_interpret(const struct lu_env *env,
3198                                  struct ptlrpc_request *req,
3199                                  struct osc_enqueue_args *aa, int rc)
3200 {
3201         struct ldlm_lock *lock;
3202         struct lustre_handle handle;
3203         __u32 mode;
3204
3205         /* Make a local copy of a lock handle and a mode, because aa->oa_*
3206          * might be freed anytime after lock upcall has been called. */
3207         lustre_handle_copy(&handle, aa->oa_lockh);
3208         mode = aa->oa_ei->ei_mode;
3209
3210         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3211          * be valid. */
3212         lock = ldlm_handle2lock(&handle);
3213
3214         /* Take an additional reference so that a blocking AST that
3215          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
3216          * to arrive after an upcall has been executed by
3217          * osc_enqueue_fini(). */
3218         ldlm_lock_addref(&handle, mode);
3219
3220         /* Complete obtaining the lock procedure. */
3221         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3222                                    mode, aa->oa_flags, aa->oa_lvb,
3223                                    sizeof(*aa->oa_lvb), &handle, rc);
3224         /* Complete osc stuff. */
3225         rc = osc_enqueue_fini(req, aa->oa_lvb,
3226                               aa->oa_upcall, aa->oa_cookie, aa->oa_flags, rc);
3227
3228         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
3229
3230         /* Release the lock for async request. */
3231         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
3232                 /*
3233                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
3234                  * not already released by
3235                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
3236                  */
3237                 ldlm_lock_decref(&handle, mode);
3238
3239         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3240                  aa->oa_lockh, req, aa);
3241         ldlm_lock_decref(&handle, mode);
3242         LDLM_LOCK_PUT(lock);
3243         return rc;
3244 }
3245
3246 void osc_update_enqueue(struct lustre_handle *lov_lockhp,
3247                         struct lov_oinfo *loi, int flags,
3248                         struct ost_lvb *lvb, __u32 mode, int rc)
3249 {
3250         if (rc == ELDLM_OK) {
3251                 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
3252                 __u64 tmp;
3253
3254                 LASSERT(lock != NULL);
3255                 loi->loi_lvb = *lvb;
3256                 tmp = loi->loi_lvb.lvb_size;
3257                 /* Extend KMS up to the end of this lock and no further
3258                  * A lock on [x,y] means a KMS of up to y + 1 bytes! */
3259                 if (tmp > lock->l_policy_data.l_extent.end)
3260                         tmp = lock->l_policy_data.l_extent.end + 1;
3261                 if (tmp >= loi->loi_kms) {
3262                         LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
3263                                    ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
3264                         loi_kms_set(loi, tmp);
3265                 } else {
3266                         LDLM_DEBUG(lock, "lock acquired, setting rss="
3267                                    LPU64"; leaving kms="LPU64", end="LPU64,
3268                                    loi->loi_lvb.lvb_size, loi->loi_kms,
3269                                    lock->l_policy_data.l_extent.end);
3270                 }
3271                 ldlm_lock_allow_match(lock);
3272                 LDLM_LOCK_PUT(lock);
3273         } else if (rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT)) {
3274                 loi->loi_lvb = *lvb;
3275                 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
3276                        " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
3277                 rc = ELDLM_OK;
3278         }
3279 }
3280 EXPORT_SYMBOL(osc_update_enqueue);
3281
3282 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
3283
3284 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3285  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3286  * other synchronous requests, however keeping some locks and trying to obtain
3287  * others may take a considerable amount of time in a case of ost failure; and
3288  * when other sync requests do not get released lock from a client, the client
3289  * is excluded from the cluster -- such scenarious make the life difficult, so
3290  * release locks just after they are obtained. */
3291 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3292                      int *flags, ldlm_policy_data_t *policy,
3293                      struct ost_lvb *lvb, int kms_valid,
3294                      obd_enqueue_update_f upcall, void *cookie,
3295                      struct ldlm_enqueue_info *einfo,
3296                      struct lustre_handle *lockh,
3297                      struct ptlrpc_request_set *rqset, int async)
3298 {
3299         struct obd_device *obd = exp->exp_obd;
3300         struct ptlrpc_request *req = NULL;
3301         int intent = *flags & LDLM_FL_HAS_INTENT;
3302         ldlm_mode_t mode;
3303         int rc;
3304         ENTRY;
3305
3306         /* Filesystem lock extents are extended to page boundaries so that
3307          * dealing with the page cache is a little smoother.  */
3308         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3309         policy->l_extent.end |= ~CFS_PAGE_MASK;
3310
3311         /*
3312          * kms is not valid when either object is completely fresh (so that no
3313          * locks are cached), or object was evicted. In the latter case cached
3314          * lock cannot be used, because it would prime inode state with
3315          * potentially stale LVB.
3316          */
3317         if (!kms_valid)
3318                 goto no_match;
3319
3320         /* Next, search for already existing extent locks that will cover us */
3321         /* If we're trying to read, we also search for an existing PW lock.  The
3322          * VFS and page cache already protect us locally, so lots of readers/
3323          * writers can share a single PW lock.
3324          *
3325          * There are problems with conversion deadlocks, so instead of
3326          * converting a read lock to a write lock, we'll just enqueue a new
3327          * one.
3328          *
3329          * At some point we should cancel the read lock instead of making them
3330          * send us a blocking callback, but there are problems with canceling
3331          * locks out from other users right now, too. */
3332         mode = einfo->ei_mode;
3333         if (einfo->ei_mode == LCK_PR)
3334                 mode |= LCK_PW;
3335         mode = ldlm_lock_match(obd->obd_namespace,
3336                                *flags | LDLM_FL_LVB_READY, res_id,
3337                                einfo->ei_type, policy, mode, lockh, 0);
3338         if (mode) {
3339                 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
3340
3341                 if (matched->l_ast_data == NULL ||
3342                     matched->l_ast_data == einfo->ei_cbdata) {
3343                         /* addref the lock only if not async requests and PW
3344                          * lock is matched whereas we asked for PR. */
3345                         if (!rqset && einfo->ei_mode != mode)
3346                                 ldlm_lock_addref(lockh, LCK_PR);
3347                         osc_set_lock_data_with_check(matched, einfo, *flags);
3348                         if (intent) {
3349                                 /* I would like to be able to ASSERT here that
3350                                  * rss <= kms, but I can't, for reasons which
3351                                  * are explained in lov_enqueue() */
3352                         }
3353
3354                         /* We already have a lock, and it's referenced */
3355                         (*upcall)(cookie, ELDLM_OK);
3356
3357                         /* For async requests, decref the lock. */
3358                         if (einfo->ei_mode != mode)
3359                                 ldlm_lock_decref(lockh, LCK_PW);
3360                         else if (rqset)
3361                                 ldlm_lock_decref(lockh, einfo->ei_mode);
3362                         LDLM_LOCK_PUT(matched);
3363                         RETURN(ELDLM_OK);
3364                 } else
3365                         ldlm_lock_decref(lockh, mode);
3366                 LDLM_LOCK_PUT(matched);
3367         }
3368
3369  no_match:
3370         if (intent) {
3371                 CFS_LIST_HEAD(cancels);
3372                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3373                                            &RQF_LDLM_ENQUEUE_LVB);
3374                 if (req == NULL)
3375                         RETURN(-ENOMEM);
3376
3377                 rc = ldlm_prep_enqueue_req(exp, req, &cancels, 0);
3378                 if (rc)
3379                         RETURN(rc);
3380
3381                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
3382                                      sizeof *lvb);
3383                 ptlrpc_request_set_replen(req);
3384         }
3385
3386         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
3387         *flags &= ~LDLM_FL_BLOCK_GRANTED;
3388
3389         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
3390                               sizeof(*lvb), lockh, async);
3391         if (rqset) {
3392                 if (!rc) {
3393                         struct osc_enqueue_args *aa;
3394                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3395                         aa = ptlrpc_req_async_args(req);
3396                         aa->oa_ei = einfo;
3397                         aa->oa_exp = exp;
3398                         aa->oa_flags  = flags;
3399                         aa->oa_upcall = upcall;
3400                         aa->oa_cookie = cookie;
3401                         aa->oa_lvb    = lvb;
3402                         aa->oa_lockh  = lockh;
3403
3404                         req->rq_interpret_reply =
3405                                 (ptlrpc_interpterer_t)osc_enqueue_interpret;
3406                         if (rqset == PTLRPCD_SET)
3407                                 ptlrpcd_add_req(req, PSCOPE_OTHER);
3408                         else
3409                                 ptlrpc_set_add_req(rqset, req);
3410                 } else if (intent) {
3411                         ptlrpc_req_finished(req);
3412                 }
3413                 RETURN(rc);
3414         }
3415
3416         rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, rc);
3417         if (intent)
3418                 ptlrpc_req_finished(req);
3419
3420         RETURN(rc);
3421 }
3422
3423 static int osc_enqueue(struct obd_export *exp, struct obd_info *oinfo,
3424                        struct ldlm_enqueue_info *einfo,
3425                        struct ptlrpc_request_set *rqset)
3426 {
3427         struct ldlm_res_id res_id;
3428         int rc;
3429         ENTRY;
3430
3431         osc_build_res_name(oinfo->oi_md->lsm_object_id,
3432                            oinfo->oi_md->lsm_object_seq, &res_id);
3433
3434         rc = osc_enqueue_base(exp, &res_id, &oinfo->oi_flags, &oinfo->oi_policy,
3435                               &oinfo->oi_md->lsm_oinfo[0]->loi_lvb,
3436                               oinfo->oi_md->lsm_oinfo[0]->loi_kms_valid,
3437                               oinfo->oi_cb_up, oinfo, einfo, oinfo->oi_lockh,
3438                               rqset, rqset != NULL);
3439         RETURN(rc);
3440 }
3441
3442 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3443                    __u32 type, ldlm_policy_data_t *policy, __u32 mode,
3444                    int *flags, void *data, struct lustre_handle *lockh,
3445                    int unref)
3446 {
3447         struct obd_device *obd = exp->exp_obd;
3448         int lflags = *flags;
3449         ldlm_mode_t rc;
3450         ENTRY;
3451
3452         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
3453                 RETURN(-EIO);
3454
3455         /* Filesystem lock extents are extended to page boundaries so that
3456          * dealing with the page cache is a little smoother */
3457         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3458         policy->l_extent.end |= ~CFS_PAGE_MASK;
3459
3460         /* Next, search for already existing extent locks that will cover us */
3461         /* If we're trying to read, we also search for an existing PW lock.  The
3462          * VFS and page cache already protect us locally, so lots of readers/
3463          * writers can share a single PW lock. */
3464         rc = mode;
3465         if (mode == LCK_PR)
3466                 rc |= LCK_PW;
3467         rc = ldlm_lock_match(obd->obd_namespace, lflags,
3468                              res_id, type, policy, rc, lockh, unref);
3469         if (rc) {
3470                 if (data != NULL)
3471                         osc_set_data_with_check(lockh, data, lflags);
3472                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
3473                         ldlm_lock_addref(lockh, LCK_PR);
3474                         ldlm_lock_decref(lockh, LCK_PW);
3475                 }
3476                 RETURN(rc);
3477         }
3478         RETURN(rc);
3479 }
3480
3481 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
3482 {
3483         ENTRY;
3484
3485         if (unlikely(mode == LCK_GROUP))
3486                 ldlm_lock_decref_and_cancel(lockh, mode);
3487         else
3488                 ldlm_lock_decref(lockh, mode);
3489
3490         RETURN(0);
3491 }
3492
3493 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
3494                       __u32 mode, struct lustre_handle *lockh)
3495 {
3496         ENTRY;
3497         RETURN(osc_cancel_base(lockh, mode));
3498 }
3499
3500 static int osc_cancel_unused(struct obd_export *exp,
3501                              struct lov_stripe_md *lsm,
3502                              ldlm_cancel_flags_t flags,
3503                              void *opaque)
3504 {
3505         struct obd_device *obd = class_exp2obd(exp);
3506         struct ldlm_res_id res_id, *resp = NULL;
3507
3508         if (lsm != NULL) {
3509                 resp = osc_build_res_name(lsm->lsm_object_id,
3510                                           lsm->lsm_object_seq, &res_id);
3511         }
3512
3513         return ldlm_cli_cancel_unused(obd->obd_namespace, resp, flags, opaque);
3514 }
3515
3516 static int osc_statfs_interpret(const struct lu_env *env,
3517                                 struct ptlrpc_request *req,
3518                                 struct osc_async_args *aa, int rc)
3519 {
3520         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
3521         struct obd_statfs *msfs;
3522         __u64 used;
3523         ENTRY;
3524
3525         if (rc == -EBADR)
3526                 /* The request has in fact never been sent
3527                  * due to issues at a higher level (LOV).
3528                  * Exit immediately since the caller is
3529                  * aware of the problem and takes care
3530                  * of the clean up */
3531                  RETURN(rc);
3532
3533         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
3534             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
3535                 GOTO(out, rc = 0);
3536
3537         if (rc != 0)
3538                 GOTO(out, rc);
3539
3540         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3541         if (msfs == NULL) {
3542                 GOTO(out, rc = -EPROTO);
3543         }
3544
3545         /* Reinitialize the RDONLY and DEGRADED flags at the client
3546          * on each statfs, so they don't stay set permanently. */
3547         cfs_spin_lock(&cli->cl_oscc.oscc_lock);
3548
3549         if (unlikely(msfs->os_state & OS_STATE_DEGRADED))
3550                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_DEGRADED;
3551         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_DEGRADED))
3552                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_DEGRADED;
3553
3554         if (unlikely(msfs->os_state & OS_STATE_READONLY))
3555                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_RDONLY;
3556         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_RDONLY))
3557                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_RDONLY;
3558
3559         /* Add a bit of hysteresis so this flag isn't continually flapping,
3560          * and ensure that new files don't get extremely fragmented due to
3561          * only a small amount of available space in the filesystem.
3562          * We want to set the NOSPC flag when there is less than ~0.1% free
3563          * and clear it when there is at least ~0.2% free space, so:
3564          *                   avail < ~0.1% max          max = avail + used
3565          *            1025 * avail < avail + used       used = blocks - free
3566          *            1024 * avail < used
3567          *            1024 * avail < blocks - free                      
3568          *                   avail < ((blocks - free) >> 10)    
3569          *
3570          * On very large disk, say 16TB 0.1% will be 16 GB. We don't want to
3571          * lose that amount of space so in those cases we report no space left
3572          * if their is less than 1 GB left.                             */
3573         used = min_t(__u64,(msfs->os_blocks - msfs->os_bfree) >> 10, 1 << 30);
3574         if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) == 0) &&
3575                      ((msfs->os_ffree < 32) || (msfs->os_bavail < used))))
3576                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_NOSPC;
3577         else if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) != 0) &&
3578                 (msfs->os_ffree > 64) && (msfs->os_bavail > (used << 1))))
3579                         cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_NOSPC;
3580
3581         cfs_spin_unlock(&cli->cl_oscc.oscc_lock);
3582
3583         *aa->aa_oi->oi_osfs = *msfs;
3584 out:
3585         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3586         RETURN(rc);
3587 }
3588
3589 static int osc_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
3590                             __u64 max_age, struct ptlrpc_request_set *rqset)
3591 {
3592         struct ptlrpc_request *req;
3593         struct osc_async_args *aa;
3594         int                    rc;
3595         ENTRY;
3596
3597         /* We could possibly pass max_age in the request (as an absolute
3598          * timestamp or a "seconds.usec ago") so the target can avoid doing
3599          * extra calls into the filesystem if that isn't necessary (e.g.
3600          * during mount that would help a bit).  Having relative timestamps
3601          * is not so great if request processing is slow, while absolute
3602          * timestamps are not ideal because they need time synchronization. */
3603         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3604         if (req == NULL)
3605                 RETURN(-ENOMEM);
3606
3607         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3608         if (rc) {
3609                 ptlrpc_request_free(req);
3610                 RETURN(rc);
3611         }
3612         ptlrpc_request_set_replen(req);
3613         req->rq_request_portal = OST_CREATE_PORTAL;
3614         ptlrpc_at_set_req_timeout(req);
3615
3616         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3617                 /* procfs requests not want stat in wait for avoid deadlock */
3618                 req->rq_no_resend = 1;
3619                 req->rq_no_delay = 1;
3620         }
3621
3622         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
3623         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3624         aa = ptlrpc_req_async_args(req);
3625         aa->aa_oi = oinfo;
3626
3627         ptlrpc_set_add_req(rqset, req);
3628         RETURN(0);
3629 }
3630
3631 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3632                       __u64 max_age, __u32 flags)
3633 {
3634         struct obd_statfs     *msfs;
3635         struct ptlrpc_request *req;
3636         struct obd_import     *imp = NULL;
3637         int rc;
3638         ENTRY;
3639
3640         /*Since the request might also come from lprocfs, so we need
3641          *sync this with client_disconnect_export Bug15684*/
3642         cfs_down_read(&obd->u.cli.cl_sem);
3643         if (obd->u.cli.cl_import)
3644                 imp = class_import_get(obd->u.cli.cl_import);
3645