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