Whamcloud - gitweb
LU-11213 mdc: add async statfs
[fs/lustre-release.git] / lustre / osc / osc_request.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32
33 #define DEBUG_SUBSYSTEM S_OSC
34
35 #include <linux/workqueue.h>
36 #include <lprocfs_status.h>
37 #include <lustre_debug.h>
38 #include <lustre_dlm.h>
39 #include <lustre_fid.h>
40 #include <lustre_ha.h>
41 #include <uapi/linux/lustre/lustre_ioctl.h>
42 #include <lustre_net.h>
43 #include <lustre_obdo.h>
44 #include <obd.h>
45 #include <obd_cksum.h>
46 #include <obd_class.h>
47 #include <lustre_osc.h>
48
49 #include "osc_internal.h"
50
51 atomic_t osc_pool_req_count;
52 unsigned int osc_reqpool_maxreqcount;
53 struct ptlrpc_request_pool *osc_rq_pool;
54
55 /* max memory used for request pool, unit is MB */
56 static unsigned int osc_reqpool_mem_max = 5;
57 module_param(osc_reqpool_mem_max, uint, 0444);
58
59 static int osc_idle_timeout = 20;
60 module_param(osc_idle_timeout, uint, 0644);
61
62 #define osc_grant_args osc_brw_async_args
63
64 struct osc_setattr_args {
65         struct obdo             *sa_oa;
66         obd_enqueue_update_f     sa_upcall;
67         void                    *sa_cookie;
68 };
69
70 struct osc_fsync_args {
71         struct osc_object       *fa_obj;
72         struct obdo             *fa_oa;
73         obd_enqueue_update_f    fa_upcall;
74         void                    *fa_cookie;
75 };
76
77 struct osc_ladvise_args {
78         struct obdo             *la_oa;
79         obd_enqueue_update_f     la_upcall;
80         void                    *la_cookie;
81 };
82
83 static void osc_release_ppga(struct brw_page **ppga, size_t count);
84 static int brw_interpret(const struct lu_env *env, struct ptlrpc_request *req,
85                          void *data, int rc);
86
87 void osc_pack_req_body(struct ptlrpc_request *req, struct obdo *oa)
88 {
89         struct ost_body *body;
90
91         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
92         LASSERT(body);
93
94         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
95 }
96
97 static int osc_getattr(const struct lu_env *env, struct obd_export *exp,
98                        struct obdo *oa)
99 {
100         struct ptlrpc_request   *req;
101         struct ost_body         *body;
102         int                      rc;
103
104         ENTRY;
105         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
106         if (req == NULL)
107                 RETURN(-ENOMEM);
108
109         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
110         if (rc) {
111                 ptlrpc_request_free(req);
112                 RETURN(rc);
113         }
114
115         osc_pack_req_body(req, oa);
116
117         ptlrpc_request_set_replen(req);
118
119         rc = ptlrpc_queue_wait(req);
120         if (rc)
121                 GOTO(out, rc);
122
123         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
124         if (body == NULL)
125                 GOTO(out, rc = -EPROTO);
126
127         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
128         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
129
130         oa->o_blksize = cli_brw_size(exp->exp_obd);
131         oa->o_valid |= OBD_MD_FLBLKSZ;
132
133         EXIT;
134 out:
135         ptlrpc_req_finished(req);
136
137         return rc;
138 }
139
140 static int osc_setattr(const struct lu_env *env, struct obd_export *exp,
141                        struct obdo *oa)
142 {
143         struct ptlrpc_request   *req;
144         struct ost_body         *body;
145         int                      rc;
146
147         ENTRY;
148         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
149
150         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
151         if (req == NULL)
152                 RETURN(-ENOMEM);
153
154         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
155         if (rc) {
156                 ptlrpc_request_free(req);
157                 RETURN(rc);
158         }
159
160         osc_pack_req_body(req, oa);
161
162         ptlrpc_request_set_replen(req);
163
164         rc = ptlrpc_queue_wait(req);
165         if (rc)
166                 GOTO(out, rc);
167
168         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
169         if (body == NULL)
170                 GOTO(out, rc = -EPROTO);
171
172         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
173
174         EXIT;
175 out:
176         ptlrpc_req_finished(req);
177
178         RETURN(rc);
179 }
180
181 static int osc_setattr_interpret(const struct lu_env *env,
182                                  struct ptlrpc_request *req, void *args, int rc)
183 {
184         struct osc_setattr_args *sa = args;
185         struct ost_body *body;
186
187         ENTRY;
188
189         if (rc != 0)
190                 GOTO(out, rc);
191
192         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
193         if (body == NULL)
194                 GOTO(out, rc = -EPROTO);
195
196         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, sa->sa_oa,
197                              &body->oa);
198 out:
199         rc = sa->sa_upcall(sa->sa_cookie, rc);
200         RETURN(rc);
201 }
202
203 int osc_setattr_async(struct obd_export *exp, struct obdo *oa,
204                       obd_enqueue_update_f upcall, void *cookie,
205                       struct ptlrpc_request_set *rqset)
206 {
207         struct ptlrpc_request   *req;
208         struct osc_setattr_args *sa;
209         int                      rc;
210
211         ENTRY;
212
213         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
214         if (req == NULL)
215                 RETURN(-ENOMEM);
216
217         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
218         if (rc) {
219                 ptlrpc_request_free(req);
220                 RETURN(rc);
221         }
222
223         osc_pack_req_body(req, oa);
224
225         ptlrpc_request_set_replen(req);
226
227         /* do mds to ost setattr asynchronously */
228         if (!rqset) {
229                 /* Do not wait for response. */
230                 ptlrpcd_add_req(req);
231         } else {
232                 req->rq_interpret_reply = osc_setattr_interpret;
233
234                 CLASSERT(sizeof(*sa) <= sizeof(req->rq_async_args));
235                 sa = ptlrpc_req_async_args(req);
236                 sa->sa_oa = oa;
237                 sa->sa_upcall = upcall;
238                 sa->sa_cookie = cookie;
239
240                 if (rqset == PTLRPCD_SET)
241                         ptlrpcd_add_req(req);
242                 else
243                         ptlrpc_set_add_req(rqset, req);
244         }
245
246         RETURN(0);
247 }
248
249 static int osc_ladvise_interpret(const struct lu_env *env,
250                                  struct ptlrpc_request *req,
251                                  void *arg, int rc)
252 {
253         struct osc_ladvise_args *la = arg;
254         struct ost_body *body;
255         ENTRY;
256
257         if (rc != 0)
258                 GOTO(out, rc);
259
260         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
261         if (body == NULL)
262                 GOTO(out, rc = -EPROTO);
263
264         *la->la_oa = body->oa;
265 out:
266         rc = la->la_upcall(la->la_cookie, rc);
267         RETURN(rc);
268 }
269
270 /**
271  * If rqset is NULL, do not wait for response. Upcall and cookie could also
272  * be NULL in this case
273  */
274 int osc_ladvise_base(struct obd_export *exp, struct obdo *oa,
275                      struct ladvise_hdr *ladvise_hdr,
276                      obd_enqueue_update_f upcall, void *cookie,
277                      struct ptlrpc_request_set *rqset)
278 {
279         struct ptlrpc_request   *req;
280         struct ost_body         *body;
281         struct osc_ladvise_args *la;
282         int                      rc;
283         struct lu_ladvise       *req_ladvise;
284         struct lu_ladvise       *ladvise = ladvise_hdr->lah_advise;
285         int                      num_advise = ladvise_hdr->lah_count;
286         struct ladvise_hdr      *req_ladvise_hdr;
287         ENTRY;
288
289         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_LADVISE);
290         if (req == NULL)
291                 RETURN(-ENOMEM);
292
293         req_capsule_set_size(&req->rq_pill, &RMF_OST_LADVISE, RCL_CLIENT,
294                              num_advise * sizeof(*ladvise));
295         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_LADVISE);
296         if (rc != 0) {
297                 ptlrpc_request_free(req);
298                 RETURN(rc);
299         }
300         req->rq_request_portal = OST_IO_PORTAL;
301         ptlrpc_at_set_req_timeout(req);
302
303         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
304         LASSERT(body);
305         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
306                              oa);
307
308         req_ladvise_hdr = req_capsule_client_get(&req->rq_pill,
309                                                  &RMF_OST_LADVISE_HDR);
310         memcpy(req_ladvise_hdr, ladvise_hdr, sizeof(*ladvise_hdr));
311
312         req_ladvise = req_capsule_client_get(&req->rq_pill, &RMF_OST_LADVISE);
313         memcpy(req_ladvise, ladvise, sizeof(*ladvise) * num_advise);
314         ptlrpc_request_set_replen(req);
315
316         if (rqset == NULL) {
317                 /* Do not wait for response. */
318                 ptlrpcd_add_req(req);
319                 RETURN(0);
320         }
321
322         req->rq_interpret_reply = osc_ladvise_interpret;
323         CLASSERT(sizeof(*la) <= sizeof(req->rq_async_args));
324         la = ptlrpc_req_async_args(req);
325         la->la_oa = oa;
326         la->la_upcall = upcall;
327         la->la_cookie = cookie;
328
329         if (rqset == PTLRPCD_SET)
330                 ptlrpcd_add_req(req);
331         else
332                 ptlrpc_set_add_req(rqset, req);
333
334         RETURN(0);
335 }
336
337 static int osc_create(const struct lu_env *env, struct obd_export *exp,
338                       struct obdo *oa)
339 {
340         struct ptlrpc_request *req;
341         struct ost_body       *body;
342         int                    rc;
343         ENTRY;
344
345         LASSERT(oa != NULL);
346         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
347         LASSERT(fid_seq_is_echo(ostid_seq(&oa->o_oi)));
348
349         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
350         if (req == NULL)
351                 GOTO(out, rc = -ENOMEM);
352
353         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
354         if (rc) {
355                 ptlrpc_request_free(req);
356                 GOTO(out, rc);
357         }
358
359         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
360         LASSERT(body);
361
362         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
363
364         ptlrpc_request_set_replen(req);
365
366         rc = ptlrpc_queue_wait(req);
367         if (rc)
368                 GOTO(out_req, rc);
369
370         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
371         if (body == NULL)
372                 GOTO(out_req, rc = -EPROTO);
373
374         CDEBUG(D_INFO, "oa flags %x\n", oa->o_flags);
375         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
376
377         oa->o_blksize = cli_brw_size(exp->exp_obd);
378         oa->o_valid |= OBD_MD_FLBLKSZ;
379
380         CDEBUG(D_HA, "transno: %lld\n",
381                lustre_msg_get_transno(req->rq_repmsg));
382 out_req:
383         ptlrpc_req_finished(req);
384 out:
385         RETURN(rc);
386 }
387
388 int osc_punch_send(struct obd_export *exp, struct obdo *oa,
389                    obd_enqueue_update_f upcall, void *cookie)
390 {
391         struct ptlrpc_request *req;
392         struct osc_setattr_args *sa;
393         struct obd_import *imp = class_exp2cliimp(exp);
394         struct ost_body *body;
395         int rc;
396
397         ENTRY;
398
399         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
400         if (req == NULL)
401                 RETURN(-ENOMEM);
402
403         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
404         if (rc < 0) {
405                 ptlrpc_request_free(req);
406                 RETURN(rc);
407         }
408
409         osc_set_io_portal(req);
410
411         ptlrpc_at_set_req_timeout(req);
412
413         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
414
415         lustre_set_wire_obdo(&imp->imp_connect_data, &body->oa, oa);
416
417         ptlrpc_request_set_replen(req);
418
419         req->rq_interpret_reply = osc_setattr_interpret;
420         CLASSERT(sizeof(*sa) <= sizeof(req->rq_async_args));
421         sa = ptlrpc_req_async_args(req);
422         sa->sa_oa = oa;
423         sa->sa_upcall = upcall;
424         sa->sa_cookie = cookie;
425
426         ptlrpcd_add_req(req);
427
428         RETURN(0);
429 }
430 EXPORT_SYMBOL(osc_punch_send);
431
432 static int osc_sync_interpret(const struct lu_env *env,
433                               struct ptlrpc_request *req, void *args, int rc)
434 {
435         struct osc_fsync_args *fa = args;
436         struct ost_body *body;
437         struct cl_attr *attr = &osc_env_info(env)->oti_attr;
438         unsigned long valid = 0;
439         struct cl_object *obj;
440         ENTRY;
441
442         if (rc != 0)
443                 GOTO(out, rc);
444
445         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
446         if (body == NULL) {
447                 CERROR("can't unpack ost_body\n");
448                 GOTO(out, rc = -EPROTO);
449         }
450
451         *fa->fa_oa = body->oa;
452         obj = osc2cl(fa->fa_obj);
453
454         /* Update osc object's blocks attribute */
455         cl_object_attr_lock(obj);
456         if (body->oa.o_valid & OBD_MD_FLBLOCKS) {
457                 attr->cat_blocks = body->oa.o_blocks;
458                 valid |= CAT_BLOCKS;
459         }
460
461         if (valid != 0)
462                 cl_object_attr_update(env, obj, attr, valid);
463         cl_object_attr_unlock(obj);
464
465 out:
466         rc = fa->fa_upcall(fa->fa_cookie, rc);
467         RETURN(rc);
468 }
469
470 int osc_sync_base(struct osc_object *obj, struct obdo *oa,
471                   obd_enqueue_update_f upcall, void *cookie,
472                   struct ptlrpc_request_set *rqset)
473 {
474         struct obd_export     *exp = osc_export(obj);
475         struct ptlrpc_request *req;
476         struct ost_body       *body;
477         struct osc_fsync_args *fa;
478         int                    rc;
479         ENTRY;
480
481         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
482         if (req == NULL)
483                 RETURN(-ENOMEM);
484
485         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
486         if (rc) {
487                 ptlrpc_request_free(req);
488                 RETURN(rc);
489         }
490
491         /* overload the size and blocks fields in the oa with start/end */
492         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
493         LASSERT(body);
494         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
495
496         ptlrpc_request_set_replen(req);
497         req->rq_interpret_reply = osc_sync_interpret;
498
499         CLASSERT(sizeof(*fa) <= sizeof(req->rq_async_args));
500         fa = ptlrpc_req_async_args(req);
501         fa->fa_obj = obj;
502         fa->fa_oa = oa;
503         fa->fa_upcall = upcall;
504         fa->fa_cookie = cookie;
505
506         if (rqset == PTLRPCD_SET)
507                 ptlrpcd_add_req(req);
508         else
509                 ptlrpc_set_add_req(rqset, req);
510
511         RETURN (0);
512 }
513
514 /* Find and cancel locally locks matched by @mode in the resource found by
515  * @objid. Found locks are added into @cancel list. Returns the amount of
516  * locks added to @cancels list. */
517 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
518                                    struct list_head *cancels,
519                                    enum ldlm_mode mode, __u64 lock_flags)
520 {
521         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
522         struct ldlm_res_id res_id;
523         struct ldlm_resource *res;
524         int count;
525         ENTRY;
526
527         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
528          * export) but disabled through procfs (flag in NS).
529          *
530          * This distinguishes from a case when ELC is not supported originally,
531          * when we still want to cancel locks in advance and just cancel them
532          * locally, without sending any RPC. */
533         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
534                 RETURN(0);
535
536         ostid_build_res_name(&oa->o_oi, &res_id);
537         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
538         if (IS_ERR(res))
539                 RETURN(0);
540
541         LDLM_RESOURCE_ADDREF(res);
542         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
543                                            lock_flags, 0, NULL);
544         LDLM_RESOURCE_DELREF(res);
545         ldlm_resource_putref(res);
546         RETURN(count);
547 }
548
549 static int osc_destroy_interpret(const struct lu_env *env,
550                                  struct ptlrpc_request *req, void *args, int rc)
551 {
552         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
553
554         atomic_dec(&cli->cl_destroy_in_flight);
555         wake_up(&cli->cl_destroy_waitq);
556
557         return 0;
558 }
559
560 static int osc_can_send_destroy(struct client_obd *cli)
561 {
562         if (atomic_inc_return(&cli->cl_destroy_in_flight) <=
563             cli->cl_max_rpcs_in_flight) {
564                 /* The destroy request can be sent */
565                 return 1;
566         }
567         if (atomic_dec_return(&cli->cl_destroy_in_flight) <
568             cli->cl_max_rpcs_in_flight) {
569                 /*
570                  * The counter has been modified between the two atomic
571                  * operations.
572                  */
573                 wake_up(&cli->cl_destroy_waitq);
574         }
575         return 0;
576 }
577
578 static int osc_destroy(const struct lu_env *env, struct obd_export *exp,
579                        struct obdo *oa)
580 {
581         struct client_obd     *cli = &exp->exp_obd->u.cli;
582         struct ptlrpc_request *req;
583         struct ost_body       *body;
584         struct list_head       cancels = LIST_HEAD_INIT(cancels);
585         int rc, count;
586         ENTRY;
587
588         if (!oa) {
589                 CDEBUG(D_INFO, "oa NULL\n");
590                 RETURN(-EINVAL);
591         }
592
593         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
594                                         LDLM_FL_DISCARD_DATA);
595
596         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
597         if (req == NULL) {
598                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
599                 RETURN(-ENOMEM);
600         }
601
602         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
603                                0, &cancels, count);
604         if (rc) {
605                 ptlrpc_request_free(req);
606                 RETURN(rc);
607         }
608
609         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
610         ptlrpc_at_set_req_timeout(req);
611
612         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
613         LASSERT(body);
614         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
615
616         ptlrpc_request_set_replen(req);
617
618         req->rq_interpret_reply = osc_destroy_interpret;
619         if (!osc_can_send_destroy(cli)) {
620                 struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
621
622                 /*
623                  * Wait until the number of on-going destroy RPCs drops
624                  * under max_rpc_in_flight
625                  */
626                 rc = l_wait_event_exclusive(cli->cl_destroy_waitq,
627                                             osc_can_send_destroy(cli), &lwi);
628                 if (rc) {
629                         ptlrpc_req_finished(req);
630                         RETURN(rc);
631                 }
632         }
633
634         /* Do not wait for response */
635         ptlrpcd_add_req(req);
636         RETURN(0);
637 }
638
639 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
640                                 long writing_bytes)
641 {
642         u64 bits = OBD_MD_FLBLOCKS | OBD_MD_FLGRANT;
643
644         LASSERT(!(oa->o_valid & bits));
645
646         oa->o_valid |= bits;
647         spin_lock(&cli->cl_loi_list_lock);
648         if (OCD_HAS_FLAG(&cli->cl_import->imp_connect_data, GRANT_PARAM))
649                 oa->o_dirty = cli->cl_dirty_grant;
650         else
651                 oa->o_dirty = cli->cl_dirty_pages << PAGE_SHIFT;
652         if (unlikely(cli->cl_dirty_pages - cli->cl_dirty_transit >
653                      cli->cl_dirty_max_pages)) {
654                 CERROR("dirty %lu - %lu > dirty_max %lu\n",
655                        cli->cl_dirty_pages, cli->cl_dirty_transit,
656                        cli->cl_dirty_max_pages);
657                 oa->o_undirty = 0;
658         } else if (unlikely(atomic_long_read(&obd_dirty_pages) -
659                             atomic_long_read(&obd_dirty_transit_pages) >
660                             (long)(obd_max_dirty_pages + 1))) {
661                 /* The atomic_read() allowing the atomic_inc() are
662                  * not covered by a lock thus they may safely race and trip
663                  * this CERROR() unless we add in a small fudge factor (+1). */
664                 CERROR("%s: dirty %ld - %ld > system dirty_max %ld\n",
665                        cli_name(cli), atomic_long_read(&obd_dirty_pages),
666                        atomic_long_read(&obd_dirty_transit_pages),
667                        obd_max_dirty_pages);
668                 oa->o_undirty = 0;
669         } else if (unlikely(cli->cl_dirty_max_pages - cli->cl_dirty_pages >
670                             0x7fffffff)) {
671                 CERROR("dirty %lu - dirty_max %lu too big???\n",
672                        cli->cl_dirty_pages, cli->cl_dirty_max_pages);
673                 oa->o_undirty = 0;
674         } else {
675                 unsigned long nrpages;
676                 unsigned long undirty;
677
678                 nrpages = cli->cl_max_pages_per_rpc;
679                 nrpages *= cli->cl_max_rpcs_in_flight + 1;
680                 nrpages = max(nrpages, cli->cl_dirty_max_pages);
681                 undirty = nrpages << PAGE_SHIFT;
682                 if (OCD_HAS_FLAG(&cli->cl_import->imp_connect_data,
683                                  GRANT_PARAM)) {
684                         int nrextents;
685
686                         /* take extent tax into account when asking for more
687                          * grant space */
688                         nrextents = (nrpages + cli->cl_max_extent_pages - 1)  /
689                                      cli->cl_max_extent_pages;
690                         undirty += nrextents * cli->cl_grant_extent_tax;
691                 }
692                 /* Do not ask for more than OBD_MAX_GRANT - a margin for server
693                  * to add extent tax, etc.
694                  */
695                 oa->o_undirty = min(undirty, OBD_MAX_GRANT &
696                                     ~(PTLRPC_MAX_BRW_SIZE * 4UL));
697         }
698         oa->o_grant = cli->cl_avail_grant + cli->cl_reserved_grant;
699         oa->o_dropped = cli->cl_lost_grant;
700         cli->cl_lost_grant = 0;
701         spin_unlock(&cli->cl_loi_list_lock);
702         CDEBUG(D_CACHE, "dirty: %llu undirty: %u dropped %u grant: %llu\n",
703                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
704 }
705
706 void osc_update_next_shrink(struct client_obd *cli)
707 {
708         cli->cl_next_shrink_grant = ktime_get_seconds() +
709                                     cli->cl_grant_shrink_interval;
710
711         CDEBUG(D_CACHE, "next time %lld to shrink grant\n",
712                cli->cl_next_shrink_grant);
713 }
714
715 static void __osc_update_grant(struct client_obd *cli, u64 grant)
716 {
717         spin_lock(&cli->cl_loi_list_lock);
718         cli->cl_avail_grant += grant;
719         spin_unlock(&cli->cl_loi_list_lock);
720 }
721
722 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
723 {
724         if (body->oa.o_valid & OBD_MD_FLGRANT) {
725                 CDEBUG(D_CACHE, "got %llu extra grant\n", body->oa.o_grant);
726                 __osc_update_grant(cli, body->oa.o_grant);
727         }
728 }
729
730 /**
731  * grant thread data for shrinking space.
732  */
733 struct grant_thread_data {
734         struct list_head        gtd_clients;
735         struct mutex            gtd_mutex;
736         unsigned long           gtd_stopped:1;
737 };
738 static struct grant_thread_data client_gtd;
739
740 static int osc_shrink_grant_interpret(const struct lu_env *env,
741                                       struct ptlrpc_request *req,
742                                       void *args, int rc)
743 {
744         struct osc_grant_args *aa = args;
745         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
746         struct ost_body *body;
747
748         if (rc != 0) {
749                 __osc_update_grant(cli, aa->aa_oa->o_grant);
750                 GOTO(out, rc);
751         }
752
753         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
754         LASSERT(body);
755         osc_update_grant(cli, body);
756 out:
757         OBD_SLAB_FREE_PTR(aa->aa_oa, osc_obdo_kmem);
758
759         return rc;
760 }
761
762 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
763 {
764         spin_lock(&cli->cl_loi_list_lock);
765         oa->o_grant = cli->cl_avail_grant / 4;
766         cli->cl_avail_grant -= oa->o_grant;
767         spin_unlock(&cli->cl_loi_list_lock);
768         if (!(oa->o_valid & OBD_MD_FLFLAGS)) {
769                 oa->o_valid |= OBD_MD_FLFLAGS;
770                 oa->o_flags = 0;
771         }
772         oa->o_flags |= OBD_FL_SHRINK_GRANT;
773         osc_update_next_shrink(cli);
774 }
775
776 /* Shrink the current grant, either from some large amount to enough for a
777  * full set of in-flight RPCs, or if we have already shrunk to that limit
778  * then to enough for a single RPC.  This avoids keeping more grant than
779  * needed, and avoids shrinking the grant piecemeal. */
780 static int osc_shrink_grant(struct client_obd *cli)
781 {
782         __u64 target_bytes = (cli->cl_max_rpcs_in_flight + 1) *
783                              (cli->cl_max_pages_per_rpc << PAGE_SHIFT);
784
785         spin_lock(&cli->cl_loi_list_lock);
786         if (cli->cl_avail_grant <= target_bytes)
787                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_SHIFT;
788         spin_unlock(&cli->cl_loi_list_lock);
789
790         return osc_shrink_grant_to_target(cli, target_bytes);
791 }
792
793 int osc_shrink_grant_to_target(struct client_obd *cli, __u64 target_bytes)
794 {
795         int                     rc = 0;
796         struct ost_body        *body;
797         ENTRY;
798
799         spin_lock(&cli->cl_loi_list_lock);
800         /* Don't shrink if we are already above or below the desired limit
801          * We don't want to shrink below a single RPC, as that will negatively
802          * impact block allocation and long-term performance. */
803         if (target_bytes < cli->cl_max_pages_per_rpc << PAGE_SHIFT)
804                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_SHIFT;
805
806         if (target_bytes >= cli->cl_avail_grant) {
807                 spin_unlock(&cli->cl_loi_list_lock);
808                 RETURN(0);
809         }
810         spin_unlock(&cli->cl_loi_list_lock);
811
812         OBD_ALLOC_PTR(body);
813         if (!body)
814                 RETURN(-ENOMEM);
815
816         osc_announce_cached(cli, &body->oa, 0);
817
818         spin_lock(&cli->cl_loi_list_lock);
819         if (target_bytes >= cli->cl_avail_grant) {
820                 /* available grant has changed since target calculation */
821                 spin_unlock(&cli->cl_loi_list_lock);
822                 GOTO(out_free, rc = 0);
823         }
824         body->oa.o_grant = cli->cl_avail_grant - target_bytes;
825         cli->cl_avail_grant = target_bytes;
826         spin_unlock(&cli->cl_loi_list_lock);
827         if (!(body->oa.o_valid & OBD_MD_FLFLAGS)) {
828                 body->oa.o_valid |= OBD_MD_FLFLAGS;
829                 body->oa.o_flags = 0;
830         }
831         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
832         osc_update_next_shrink(cli);
833
834         rc = osc_set_info_async(NULL, cli->cl_import->imp_obd->obd_self_export,
835                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
836                                 sizeof(*body), body, NULL);
837         if (rc != 0)
838                 __osc_update_grant(cli, body->oa.o_grant);
839 out_free:
840         OBD_FREE_PTR(body);
841         RETURN(rc);
842 }
843
844 static int osc_should_shrink_grant(struct client_obd *client)
845 {
846         time64_t next_shrink = client->cl_next_shrink_grant;
847
848         if (client->cl_import == NULL)
849                 return 0;
850
851         if ((client->cl_import->imp_connect_data.ocd_connect_flags &
852              OBD_CONNECT_GRANT_SHRINK) == 0)
853                 return 0;
854
855         if (ktime_get_seconds() >= next_shrink - 5) {
856                 /* Get the current RPC size directly, instead of going via:
857                  * cli_brw_size(obd->u.cli.cl_import->imp_obd->obd_self_export)
858                  * Keep comment here so that it can be found by searching. */
859                 int brw_size = client->cl_max_pages_per_rpc << PAGE_SHIFT;
860
861                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
862                     client->cl_avail_grant > brw_size)
863                         return 1;
864                 else
865                         osc_update_next_shrink(client);
866         }
867         return 0;
868 }
869
870 #define GRANT_SHRINK_RPC_BATCH  100
871
872 static struct delayed_work work;
873
874 static void osc_grant_work_handler(struct work_struct *data)
875 {
876         struct client_obd *cli;
877         int rpc_sent;
878         bool init_next_shrink = true;
879         time64_t next_shrink = ktime_get_seconds() + GRANT_SHRINK_INTERVAL;
880
881         rpc_sent = 0;
882         mutex_lock(&client_gtd.gtd_mutex);
883         list_for_each_entry(cli, &client_gtd.gtd_clients,
884                             cl_grant_chain) {
885                 if (rpc_sent < GRANT_SHRINK_RPC_BATCH &&
886                     osc_should_shrink_grant(cli)) {
887                         osc_shrink_grant(cli);
888                         rpc_sent++;
889                 }
890
891                 if (!init_next_shrink) {
892                         if (cli->cl_next_shrink_grant < next_shrink &&
893                             cli->cl_next_shrink_grant > ktime_get_seconds())
894                                 next_shrink = cli->cl_next_shrink_grant;
895                 } else {
896                         init_next_shrink = false;
897                         next_shrink = cli->cl_next_shrink_grant;
898                 }
899         }
900         mutex_unlock(&client_gtd.gtd_mutex);
901
902         if (client_gtd.gtd_stopped == 1)
903                 return;
904
905         if (next_shrink > ktime_get_seconds())
906                 schedule_delayed_work(&work, msecs_to_jiffies(
907                                         (next_shrink - ktime_get_seconds()) *
908                                         MSEC_PER_SEC));
909         else
910                 schedule_work(&work.work);
911 }
912
913 void osc_schedule_grant_work(void)
914 {
915         cancel_delayed_work_sync(&work);
916         schedule_work(&work.work);
917 }
918
919 /**
920  * Start grant thread for returing grant to server for idle clients.
921  */
922 static int osc_start_grant_work(void)
923 {
924         client_gtd.gtd_stopped = 0;
925         mutex_init(&client_gtd.gtd_mutex);
926         INIT_LIST_HEAD(&client_gtd.gtd_clients);
927
928         INIT_DELAYED_WORK(&work, osc_grant_work_handler);
929         schedule_work(&work.work);
930
931         return 0;
932 }
933
934 static void osc_stop_grant_work(void)
935 {
936         client_gtd.gtd_stopped = 1;
937         cancel_delayed_work_sync(&work);
938 }
939
940 static void osc_add_grant_list(struct client_obd *client)
941 {
942         mutex_lock(&client_gtd.gtd_mutex);
943         list_add(&client->cl_grant_chain, &client_gtd.gtd_clients);
944         mutex_unlock(&client_gtd.gtd_mutex);
945 }
946
947 static void osc_del_grant_list(struct client_obd *client)
948 {
949         if (list_empty(&client->cl_grant_chain))
950                 return;
951
952         mutex_lock(&client_gtd.gtd_mutex);
953         list_del_init(&client->cl_grant_chain);
954         mutex_unlock(&client_gtd.gtd_mutex);
955 }
956
957 void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
958 {
959         /*
960          * ocd_grant is the total grant amount we're expect to hold: if we've
961          * been evicted, it's the new avail_grant amount, cl_dirty_pages will
962          * drop to 0 as inflight RPCs fail out; otherwise, it's avail_grant +
963          * dirty.
964          *
965          * race is tolerable here: if we're evicted, but imp_state already
966          * left EVICTED state, then cl_dirty_pages must be 0 already.
967          */
968         spin_lock(&cli->cl_loi_list_lock);
969         cli->cl_avail_grant = ocd->ocd_grant;
970         if (cli->cl_import->imp_state != LUSTRE_IMP_EVICTED) {
971                 cli->cl_avail_grant -= cli->cl_reserved_grant;
972                 if (OCD_HAS_FLAG(ocd, GRANT_PARAM))
973                         cli->cl_avail_grant -= cli->cl_dirty_grant;
974                 else
975                         cli->cl_avail_grant -=
976                                         cli->cl_dirty_pages << PAGE_SHIFT;
977         }
978
979         if (OCD_HAS_FLAG(ocd, GRANT_PARAM)) {
980                 u64 size;
981                 int chunk_mask;
982
983                 /* overhead for each extent insertion */
984                 cli->cl_grant_extent_tax = ocd->ocd_grant_tax_kb << 10;
985                 /* determine the appropriate chunk size used by osc_extent. */
986                 cli->cl_chunkbits = max_t(int, PAGE_SHIFT,
987                                           ocd->ocd_grant_blkbits);
988                 /* max_pages_per_rpc must be chunk aligned */
989                 chunk_mask = ~((1 << (cli->cl_chunkbits - PAGE_SHIFT)) - 1);
990                 cli->cl_max_pages_per_rpc = (cli->cl_max_pages_per_rpc +
991                                              ~chunk_mask) & chunk_mask;
992                 /* determine maximum extent size, in #pages */
993                 size = (u64)ocd->ocd_grant_max_blks << ocd->ocd_grant_blkbits;
994                 cli->cl_max_extent_pages = size >> PAGE_SHIFT;
995                 if (cli->cl_max_extent_pages == 0)
996                         cli->cl_max_extent_pages = 1;
997         } else {
998                 cli->cl_grant_extent_tax = 0;
999                 cli->cl_chunkbits = PAGE_SHIFT;
1000                 cli->cl_max_extent_pages = DT_MAX_BRW_PAGES;
1001         }
1002         spin_unlock(&cli->cl_loi_list_lock);
1003
1004         CDEBUG(D_CACHE, "%s, setting cl_avail_grant: %ld cl_lost_grant: %ld."
1005                 "chunk bits: %d cl_max_extent_pages: %d\n",
1006                 cli_name(cli),
1007                 cli->cl_avail_grant, cli->cl_lost_grant, cli->cl_chunkbits,
1008                 cli->cl_max_extent_pages);
1009
1010         if (OCD_HAS_FLAG(ocd, GRANT_SHRINK) && list_empty(&cli->cl_grant_chain))
1011                 osc_add_grant_list(cli);
1012 }
1013 EXPORT_SYMBOL(osc_init_grant);
1014
1015 /* We assume that the reason this OSC got a short read is because it read
1016  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1017  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1018  * this stripe never got written at or beyond this stripe offset yet. */
1019 static void handle_short_read(int nob_read, size_t page_count,
1020                               struct brw_page **pga)
1021 {
1022         char *ptr;
1023         int i = 0;
1024
1025         /* skip bytes read OK */
1026         while (nob_read > 0) {
1027                 LASSERT (page_count > 0);
1028
1029                 if (pga[i]->count > nob_read) {
1030                         /* EOF inside this page */
1031                         ptr = kmap(pga[i]->pg) +
1032                                 (pga[i]->off & ~PAGE_MASK);
1033                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1034                         kunmap(pga[i]->pg);
1035                         page_count--;
1036                         i++;
1037                         break;
1038                 }
1039
1040                 nob_read -= pga[i]->count;
1041                 page_count--;
1042                 i++;
1043         }
1044
1045         /* zero remaining pages */
1046         while (page_count-- > 0) {
1047                 ptr = kmap(pga[i]->pg) + (pga[i]->off & ~PAGE_MASK);
1048                 memset(ptr, 0, pga[i]->count);
1049                 kunmap(pga[i]->pg);
1050                 i++;
1051         }
1052 }
1053
1054 static int check_write_rcs(struct ptlrpc_request *req,
1055                            int requested_nob, int niocount,
1056                            size_t page_count, struct brw_page **pga)
1057 {
1058         int     i;
1059         __u32   *remote_rcs;
1060
1061         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1062                                                   sizeof(*remote_rcs) *
1063                                                   niocount);
1064         if (remote_rcs == NULL) {
1065                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1066                 return(-EPROTO);
1067         }
1068
1069         /* return error if any niobuf was in error */
1070         for (i = 0; i < niocount; i++) {
1071                 if ((int)remote_rcs[i] < 0)
1072                         return(remote_rcs[i]);
1073
1074                 if (remote_rcs[i] != 0) {
1075                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1076                                 i, remote_rcs[i], req);
1077                         return(-EPROTO);
1078                 }
1079         }
1080         if (req->rq_bulk != NULL &&
1081             req->rq_bulk->bd_nob_transferred != requested_nob) {
1082                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1083                        req->rq_bulk->bd_nob_transferred, requested_nob);
1084                 return(-EPROTO);
1085         }
1086
1087         return (0);
1088 }
1089
1090 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1091 {
1092         if (p1->flag != p2->flag) {
1093                 unsigned mask = ~(OBD_BRW_FROM_GRANT | OBD_BRW_NOCACHE |
1094                                   OBD_BRW_SYNC       | OBD_BRW_ASYNC   |
1095                                   OBD_BRW_NOQUOTA    | OBD_BRW_SOFT_SYNC);
1096
1097                 /* warn if we try to combine flags that we don't know to be
1098                  * safe to combine */
1099                 if (unlikely((p1->flag & mask) != (p2->flag & mask))) {
1100                         CWARN("Saw flags 0x%x and 0x%x in the same brw, please "
1101                               "report this at https://jira.whamcloud.com/\n",
1102                               p1->flag, p2->flag);
1103                 }
1104                 return 0;
1105         }
1106
1107         return (p1->off + p1->count == p2->off);
1108 }
1109
1110 #if IS_ENABLED(CONFIG_CRC_T10DIF)
1111 static int osc_checksum_bulk_t10pi(const char *obd_name, int nob,
1112                                    size_t pg_count, struct brw_page **pga,
1113                                    int opc, obd_dif_csum_fn *fn,
1114                                    int sector_size,
1115                                    u32 *check_sum)
1116 {
1117         struct ahash_request *req;
1118         /* Used Adler as the default checksum type on top of DIF tags */
1119         unsigned char cfs_alg = cksum_obd2cfs(OBD_CKSUM_T10_TOP);
1120         struct page *__page;
1121         unsigned char *buffer;
1122         __u16 *guard_start;
1123         unsigned int bufsize;
1124         int guard_number;
1125         int used_number = 0;
1126         int used;
1127         u32 cksum;
1128         int rc = 0;
1129         int i = 0;
1130
1131         LASSERT(pg_count > 0);
1132
1133         __page = alloc_page(GFP_KERNEL);
1134         if (__page == NULL)
1135                 return -ENOMEM;
1136
1137         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1138         if (IS_ERR(req)) {
1139                 rc = PTR_ERR(req);
1140                 CERROR("%s: unable to initialize checksum hash %s: rc = %d\n",
1141                        obd_name, cfs_crypto_hash_name(cfs_alg), rc);
1142                 GOTO(out, rc);
1143         }
1144
1145         buffer = kmap(__page);
1146         guard_start = (__u16 *)buffer;
1147         guard_number = PAGE_SIZE / sizeof(*guard_start);
1148         while (nob > 0 && pg_count > 0) {
1149                 unsigned int count = pga[i]->count > nob ? nob : pga[i]->count;
1150
1151                 /* corrupt the data before we compute the checksum, to
1152                  * simulate an OST->client data error */
1153                 if (unlikely(i == 0 && opc == OST_READ &&
1154                              OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))) {
1155                         unsigned char *ptr = kmap(pga[i]->pg);
1156                         int off = pga[i]->off & ~PAGE_MASK;
1157
1158                         memcpy(ptr + off, "bad1", min_t(typeof(nob), 4, nob));
1159                         kunmap(pga[i]->pg);
1160                 }
1161
1162                 /*
1163                  * The left guard number should be able to hold checksums of a
1164                  * whole page
1165                  */
1166                 rc = obd_page_dif_generate_buffer(obd_name, pga[i]->pg,
1167                                                   pga[i]->off & ~PAGE_MASK,
1168                                                   count,
1169                                                   guard_start + used_number,
1170                                                   guard_number - used_number,
1171                                                   &used, sector_size,
1172                                                   fn);
1173                 if (rc)
1174                         break;
1175
1176                 used_number += used;
1177                 if (used_number == guard_number) {
1178                         cfs_crypto_hash_update_page(req, __page, 0,
1179                                 used_number * sizeof(*guard_start));
1180                         used_number = 0;
1181                 }
1182
1183                 nob -= pga[i]->count;
1184                 pg_count--;
1185                 i++;
1186         }
1187         kunmap(__page);
1188         if (rc)
1189                 GOTO(out, rc);
1190
1191         if (used_number != 0)
1192                 cfs_crypto_hash_update_page(req, __page, 0,
1193                         used_number * sizeof(*guard_start));
1194
1195         bufsize = sizeof(cksum);
1196         cfs_crypto_hash_final(req, (unsigned char *)&cksum, &bufsize);
1197
1198         /* For sending we only compute the wrong checksum instead
1199          * of corrupting the data so it is still correct on a redo */
1200         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1201                 cksum++;
1202
1203         *check_sum = cksum;
1204 out:
1205         __free_page(__page);
1206         return rc;
1207 }
1208 #else /* !CONFIG_CRC_T10DIF */
1209 #define obd_dif_ip_fn NULL
1210 #define obd_dif_crc_fn NULL
1211 #define osc_checksum_bulk_t10pi(name, nob, pgc, pga, opc, fn, ssize, csum)  \
1212         -EOPNOTSUPP
1213 #endif /* CONFIG_CRC_T10DIF */
1214
1215 static int osc_checksum_bulk(int nob, size_t pg_count,
1216                              struct brw_page **pga, int opc,
1217                              enum cksum_types cksum_type,
1218                              u32 *cksum)
1219 {
1220         int                             i = 0;
1221         struct ahash_request           *req;
1222         unsigned int                    bufsize;
1223         unsigned char                   cfs_alg = cksum_obd2cfs(cksum_type);
1224
1225         LASSERT(pg_count > 0);
1226
1227         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1228         if (IS_ERR(req)) {
1229                 CERROR("Unable to initialize checksum hash %s\n",
1230                        cfs_crypto_hash_name(cfs_alg));
1231                 return PTR_ERR(req);
1232         }
1233
1234         while (nob > 0 && pg_count > 0) {
1235                 unsigned int count = pga[i]->count > nob ? nob : pga[i]->count;
1236
1237                 /* corrupt the data before we compute the checksum, to
1238                  * simulate an OST->client data error */
1239                 if (i == 0 && opc == OST_READ &&
1240                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE)) {
1241                         unsigned char *ptr = kmap(pga[i]->pg);
1242                         int off = pga[i]->off & ~PAGE_MASK;
1243
1244                         memcpy(ptr + off, "bad1", min_t(typeof(nob), 4, nob));
1245                         kunmap(pga[i]->pg);
1246                 }
1247                 cfs_crypto_hash_update_page(req, pga[i]->pg,
1248                                             pga[i]->off & ~PAGE_MASK,
1249                                             count);
1250                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d\n",
1251                                (int)(pga[i]->off & ~PAGE_MASK));
1252
1253                 nob -= pga[i]->count;
1254                 pg_count--;
1255                 i++;
1256         }
1257
1258         bufsize = sizeof(*cksum);
1259         cfs_crypto_hash_final(req, (unsigned char *)cksum, &bufsize);
1260
1261         /* For sending we only compute the wrong checksum instead
1262          * of corrupting the data so it is still correct on a redo */
1263         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1264                 (*cksum)++;
1265
1266         return 0;
1267 }
1268
1269 static int osc_checksum_bulk_rw(const char *obd_name,
1270                                 enum cksum_types cksum_type,
1271                                 int nob, size_t pg_count,
1272                                 struct brw_page **pga, int opc,
1273                                 u32 *check_sum)
1274 {
1275         obd_dif_csum_fn *fn = NULL;
1276         int sector_size = 0;
1277         int rc;
1278
1279         ENTRY;
1280         obd_t10_cksum2dif(cksum_type, &fn, &sector_size);
1281
1282         if (fn)
1283                 rc = osc_checksum_bulk_t10pi(obd_name, nob, pg_count, pga,
1284                                              opc, fn, sector_size, check_sum);
1285         else
1286                 rc = osc_checksum_bulk(nob, pg_count, pga, opc, cksum_type,
1287                                        check_sum);
1288
1289         RETURN(rc);
1290 }
1291
1292 static int
1293 osc_brw_prep_request(int cmd, struct client_obd *cli, struct obdo *oa,
1294                      u32 page_count, struct brw_page **pga,
1295                      struct ptlrpc_request **reqp, int resend)
1296 {
1297         struct ptlrpc_request   *req;
1298         struct ptlrpc_bulk_desc *desc;
1299         struct ost_body         *body;
1300         struct obd_ioobj        *ioobj;
1301         struct niobuf_remote    *niobuf;
1302         int niocount, i, requested_nob, opc, rc, short_io_size = 0;
1303         struct osc_brw_async_args *aa;
1304         struct req_capsule      *pill;
1305         struct brw_page *pg_prev;
1306         void *short_io_buf;
1307         const char *obd_name = cli->cl_import->imp_obd->obd_name;
1308
1309         ENTRY;
1310         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1311                 RETURN(-ENOMEM); /* Recoverable */
1312         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1313                 RETURN(-EINVAL); /* Fatal */
1314
1315         if ((cmd & OBD_BRW_WRITE) != 0) {
1316                 opc = OST_WRITE;
1317                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1318                                                 osc_rq_pool,
1319                                                 &RQF_OST_BRW_WRITE);
1320         } else {
1321                 opc = OST_READ;
1322                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW_READ);
1323         }
1324         if (req == NULL)
1325                 RETURN(-ENOMEM);
1326
1327         for (niocount = i = 1; i < page_count; i++) {
1328                 if (!can_merge_pages(pga[i - 1], pga[i]))
1329                         niocount++;
1330         }
1331
1332         pill = &req->rq_pill;
1333         req_capsule_set_size(pill, &RMF_OBD_IOOBJ, RCL_CLIENT,
1334                              sizeof(*ioobj));
1335         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1336                              niocount * sizeof(*niobuf));
1337
1338         for (i = 0; i < page_count; i++)
1339                 short_io_size += pga[i]->count;
1340
1341         /* Check if read/write is small enough to be a short io. */
1342         if (short_io_size > cli->cl_max_short_io_bytes || niocount > 1 ||
1343             !imp_connect_shortio(cli->cl_import))
1344                 short_io_size = 0;
1345
1346         req_capsule_set_size(pill, &RMF_SHORT_IO, RCL_CLIENT,
1347                              opc == OST_READ ? 0 : short_io_size);
1348         if (opc == OST_READ)
1349                 req_capsule_set_size(pill, &RMF_SHORT_IO, RCL_SERVER,
1350                                      short_io_size);
1351
1352         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1353         if (rc) {
1354                 ptlrpc_request_free(req);
1355                 RETURN(rc);
1356         }
1357         osc_set_io_portal(req);
1358
1359         ptlrpc_at_set_req_timeout(req);
1360         /* ask ptlrpc not to resend on EINPROGRESS since BRWs have their own
1361          * retry logic */
1362         req->rq_no_retry_einprogress = 1;
1363
1364         if (short_io_size != 0) {
1365                 desc = NULL;
1366                 short_io_buf = NULL;
1367                 goto no_bulk;
1368         }
1369
1370         desc = ptlrpc_prep_bulk_imp(req, page_count,
1371                 cli->cl_import->imp_connect_data.ocd_brw_size >> LNET_MTU_BITS,
1372                 (opc == OST_WRITE ? PTLRPC_BULK_GET_SOURCE :
1373                         PTLRPC_BULK_PUT_SINK) |
1374                         PTLRPC_BULK_BUF_KIOV,
1375                 OST_BULK_PORTAL,
1376                 &ptlrpc_bulk_kiov_pin_ops);
1377
1378         if (desc == NULL)
1379                 GOTO(out, rc = -ENOMEM);
1380         /* NB request now owns desc and will free it when it gets freed */
1381 no_bulk:
1382         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1383         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1384         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1385         LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
1386
1387         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
1388
1389         /* For READ and WRITE, we can't fill o_uid and o_gid using from_kuid()
1390          * and from_kgid(), because they are asynchronous. Fortunately, variable
1391          * oa contains valid o_uid and o_gid in these two operations.
1392          * Besides, filling o_uid and o_gid is enough for nrs-tbf, see LU-9658.
1393          * OBD_MD_FLUID and OBD_MD_FLUID is not set in order to avoid breaking
1394          * other process logic */
1395         body->oa.o_uid = oa->o_uid;
1396         body->oa.o_gid = oa->o_gid;
1397
1398         obdo_to_ioobj(oa, ioobj);
1399         ioobj->ioo_bufcnt = niocount;
1400         /* The high bits of ioo_max_brw tells server _maximum_ number of bulks
1401          * that might be send for this request.  The actual number is decided
1402          * when the RPC is finally sent in ptlrpc_register_bulk(). It sends
1403          * "max - 1" for old client compatibility sending "0", and also so the
1404          * the actual maximum is a power-of-two number, not one less. LU-1431 */
1405         if (desc != NULL)
1406                 ioobj_max_brw_set(ioobj, desc->bd_md_max_brw);
1407         else /* short io */
1408                 ioobj_max_brw_set(ioobj, 0);
1409
1410         if (short_io_size != 0) {
1411                 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1412                         body->oa.o_valid |= OBD_MD_FLFLAGS;
1413                         body->oa.o_flags = 0;
1414                 }
1415                 body->oa.o_flags |= OBD_FL_SHORT_IO;
1416                 CDEBUG(D_CACHE, "Using short io for data transfer, size = %d\n",
1417                        short_io_size);
1418                 if (opc == OST_WRITE) {
1419                         short_io_buf = req_capsule_client_get(pill,
1420                                                               &RMF_SHORT_IO);
1421                         LASSERT(short_io_buf != NULL);
1422                 }
1423         }
1424
1425         LASSERT(page_count > 0);
1426         pg_prev = pga[0];
1427         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1428                 struct brw_page *pg = pga[i];
1429                 int poff = pg->off & ~PAGE_MASK;
1430
1431                 LASSERT(pg->count > 0);
1432                 /* make sure there is no gap in the middle of page array */
1433                 LASSERTF(page_count == 1 ||
1434                          (ergo(i == 0, poff + pg->count == PAGE_SIZE) &&
1435                           ergo(i > 0 && i < page_count - 1,
1436                                poff == 0 && pg->count == PAGE_SIZE)   &&
1437                           ergo(i == page_count - 1, poff == 0)),
1438                          "i: %d/%d pg: %p off: %llu, count: %u\n",
1439                          i, page_count, pg, pg->off, pg->count);
1440                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1441                          "i %d p_c %u pg %p [pri %lu ind %lu] off %llu"
1442                          " prev_pg %p [pri %lu ind %lu] off %llu\n",
1443                          i, page_count,
1444                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1445                          pg_prev->pg, page_private(pg_prev->pg),
1446                          pg_prev->pg->index, pg_prev->off);
1447                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1448                         (pg->flag & OBD_BRW_SRVLOCK));
1449                 if (short_io_size != 0 && opc == OST_WRITE) {
1450                         unsigned char *ptr = ll_kmap_atomic(pg->pg, KM_USER0);
1451
1452                         LASSERT(short_io_size >= requested_nob + pg->count);
1453                         memcpy(short_io_buf + requested_nob,
1454                                ptr + poff,
1455                                pg->count);
1456                         ll_kunmap_atomic(ptr, KM_USER0);
1457                 } else if (short_io_size == 0) {
1458                         desc->bd_frag_ops->add_kiov_frag(desc, pg->pg, poff,
1459                                                          pg->count);
1460                 }
1461                 requested_nob += pg->count;
1462
1463                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1464                         niobuf--;
1465                         niobuf->rnb_len += pg->count;
1466                 } else {
1467                         niobuf->rnb_offset = pg->off;
1468                         niobuf->rnb_len    = pg->count;
1469                         niobuf->rnb_flags  = pg->flag;
1470                 }
1471                 pg_prev = pg;
1472         }
1473
1474         LASSERTF((void *)(niobuf - niocount) ==
1475                 req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE),
1476                 "want %p - real %p\n", req_capsule_client_get(&req->rq_pill,
1477                 &RMF_NIOBUF_REMOTE), (void *)(niobuf - niocount));
1478
1479         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1480         if (resend) {
1481                 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1482                         body->oa.o_valid |= OBD_MD_FLFLAGS;
1483                         body->oa.o_flags = 0;
1484                 }
1485                 body->oa.o_flags |= OBD_FL_RECOV_RESEND;
1486         }
1487
1488         if (osc_should_shrink_grant(cli))
1489                 osc_shrink_grant_local(cli, &body->oa);
1490
1491         /* size[REQ_REC_OFF] still sizeof (*body) */
1492         if (opc == OST_WRITE) {
1493                 if (cli->cl_checksum &&
1494                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1495                         /* store cl_cksum_type in a local variable since
1496                          * it can be changed via lprocfs */
1497                         enum cksum_types cksum_type = cli->cl_cksum_type;
1498
1499                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1500                                 body->oa.o_flags = 0;
1501
1502                         body->oa.o_flags |= obd_cksum_type_pack(obd_name,
1503                                                                 cksum_type);
1504                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1505
1506                         rc = osc_checksum_bulk_rw(obd_name, cksum_type,
1507                                                   requested_nob, page_count,
1508                                                   pga, OST_WRITE,
1509                                                   &body->oa.o_cksum);
1510                         if (rc < 0) {
1511                                 CDEBUG(D_PAGE, "failed to checksum, rc = %d\n",
1512                                        rc);
1513                                 GOTO(out, rc);
1514                         }
1515                         CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1516                                body->oa.o_cksum);
1517
1518                         /* save this in 'oa', too, for later checking */
1519                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1520                         oa->o_flags |= obd_cksum_type_pack(obd_name,
1521                                                            cksum_type);
1522                 } else {
1523                         /* clear out the checksum flag, in case this is a
1524                          * resend but cl_checksum is no longer set. b=11238 */
1525                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1526                 }
1527                 oa->o_cksum = body->oa.o_cksum;
1528                 /* 1 RC per niobuf */
1529                 req_capsule_set_size(pill, &RMF_RCS, RCL_SERVER,
1530                                      sizeof(__u32) * niocount);
1531         } else {
1532                 if (cli->cl_checksum &&
1533                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1534                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1535                                 body->oa.o_flags = 0;
1536                         body->oa.o_flags |= obd_cksum_type_pack(obd_name,
1537                                 cli->cl_cksum_type);
1538                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1539                 }
1540
1541                 /* Client cksum has been already copied to wire obdo in previous
1542                  * lustre_set_wire_obdo(), and in the case a bulk-read is being
1543                  * resent due to cksum error, this will allow Server to
1544                  * check+dump pages on its side */
1545         }
1546         ptlrpc_request_set_replen(req);
1547
1548         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1549         aa = ptlrpc_req_async_args(req);
1550         aa->aa_oa = oa;
1551         aa->aa_requested_nob = requested_nob;
1552         aa->aa_nio_count = niocount;
1553         aa->aa_page_count = page_count;
1554         aa->aa_resends = 0;
1555         aa->aa_ppga = pga;
1556         aa->aa_cli = cli;
1557         INIT_LIST_HEAD(&aa->aa_oaps);
1558
1559         *reqp = req;
1560         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1561         CDEBUG(D_RPCTRACE, "brw rpc %p - object "DOSTID" offset %lld<>%lld\n",
1562                 req, POSTID(&oa->o_oi), niobuf[0].rnb_offset,
1563                 niobuf[niocount - 1].rnb_offset + niobuf[niocount - 1].rnb_len);
1564         RETURN(0);
1565
1566  out:
1567         ptlrpc_req_finished(req);
1568         RETURN(rc);
1569 }
1570
1571 char dbgcksum_file_name[PATH_MAX];
1572
1573 static void dump_all_bulk_pages(struct obdo *oa, __u32 page_count,
1574                                 struct brw_page **pga, __u32 server_cksum,
1575                                 __u32 client_cksum)
1576 {
1577         struct file *filp;
1578         int rc, i;
1579         unsigned int len;
1580         char *buf;
1581
1582         /* will only keep dump of pages on first error for the same range in
1583          * file/fid, not during the resends/retries. */
1584         snprintf(dbgcksum_file_name, sizeof(dbgcksum_file_name),
1585                  "%s-checksum_dump-osc-"DFID":[%llu-%llu]-%x-%x",
1586                  (strncmp(libcfs_debug_file_path_arr, "NONE", 4) != 0 ?
1587                   libcfs_debug_file_path_arr :
1588                   LIBCFS_DEBUG_FILE_PATH_DEFAULT),
1589                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : 0ULL,
1590                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1591                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1592                  pga[0]->off,
1593                  pga[page_count-1]->off + pga[page_count-1]->count - 1,
1594                  client_cksum, server_cksum);
1595         filp = filp_open(dbgcksum_file_name,
1596                          O_CREAT | O_EXCL | O_WRONLY | O_LARGEFILE, 0600);
1597         if (IS_ERR(filp)) {
1598                 rc = PTR_ERR(filp);
1599                 if (rc == -EEXIST)
1600                         CDEBUG(D_INFO, "%s: can't open to dump pages with "
1601                                "checksum error: rc = %d\n", dbgcksum_file_name,
1602                                rc);
1603                 else
1604                         CERROR("%s: can't open to dump pages with checksum "
1605                                "error: rc = %d\n", dbgcksum_file_name, rc);
1606                 return;
1607         }
1608
1609         for (i = 0; i < page_count; i++) {
1610                 len = pga[i]->count;
1611                 buf = kmap(pga[i]->pg);
1612                 while (len != 0) {
1613                         rc = cfs_kernel_write(filp, buf, len, &filp->f_pos);
1614                         if (rc < 0) {
1615                                 CERROR("%s: wanted to write %u but got %d "
1616                                        "error\n", dbgcksum_file_name, len, rc);
1617                                 break;
1618                         }
1619                         len -= rc;
1620                         buf += rc;
1621                         CDEBUG(D_INFO, "%s: wrote %d bytes\n",
1622                                dbgcksum_file_name, rc);
1623                 }
1624                 kunmap(pga[i]->pg);
1625         }
1626
1627         rc = ll_vfs_fsync_range(filp, 0, LLONG_MAX, 1);
1628         if (rc)
1629                 CERROR("%s: sync returns %d\n", dbgcksum_file_name, rc);
1630         filp_close(filp, NULL);
1631         return;
1632 }
1633
1634 static int
1635 check_write_checksum(struct obdo *oa, const struct lnet_process_id *peer,
1636                      __u32 client_cksum, __u32 server_cksum,
1637                      struct osc_brw_async_args *aa)
1638 {
1639         const char *obd_name = aa->aa_cli->cl_import->imp_obd->obd_name;
1640         enum cksum_types cksum_type;
1641         obd_dif_csum_fn *fn = NULL;
1642         int sector_size = 0;
1643         __u32 new_cksum;
1644         char *msg;
1645         int rc;
1646
1647         if (server_cksum == client_cksum) {
1648                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1649                 return 0;
1650         }
1651
1652         if (aa->aa_cli->cl_checksum_dump)
1653                 dump_all_bulk_pages(oa, aa->aa_page_count, aa->aa_ppga,
1654                                     server_cksum, client_cksum);
1655
1656         cksum_type = obd_cksum_type_unpack(oa->o_valid & OBD_MD_FLFLAGS ?
1657                                            oa->o_flags : 0);
1658
1659         switch (cksum_type) {
1660         case OBD_CKSUM_T10IP512:
1661                 fn = obd_dif_ip_fn;
1662                 sector_size = 512;
1663                 break;
1664         case OBD_CKSUM_T10IP4K:
1665                 fn = obd_dif_ip_fn;
1666                 sector_size = 4096;
1667                 break;
1668         case OBD_CKSUM_T10CRC512:
1669                 fn = obd_dif_crc_fn;
1670                 sector_size = 512;
1671                 break;
1672         case OBD_CKSUM_T10CRC4K:
1673                 fn = obd_dif_crc_fn;
1674                 sector_size = 4096;
1675                 break;
1676         default:
1677                 break;
1678         }
1679
1680         if (fn)
1681                 rc = osc_checksum_bulk_t10pi(obd_name, aa->aa_requested_nob,
1682                                              aa->aa_page_count, aa->aa_ppga,
1683                                              OST_WRITE, fn, sector_size,
1684                                              &new_cksum);
1685         else
1686                 rc = osc_checksum_bulk(aa->aa_requested_nob, aa->aa_page_count,
1687                                        aa->aa_ppga, OST_WRITE, cksum_type,
1688                                        &new_cksum);
1689
1690         if (rc < 0)
1691                 msg = "failed to calculate the client write checksum";
1692         else if (cksum_type != obd_cksum_type_unpack(aa->aa_oa->o_flags))
1693                 msg = "the server did not use the checksum type specified in "
1694                       "the original request - likely a protocol problem";
1695         else if (new_cksum == server_cksum)
1696                 msg = "changed on the client after we checksummed it - "
1697                       "likely false positive due to mmap IO (bug 11742)";
1698         else if (new_cksum == client_cksum)
1699                 msg = "changed in transit before arrival at OST";
1700         else
1701                 msg = "changed in transit AND doesn't match the original - "
1702                       "likely false positive due to mmap IO (bug 11742)";
1703
1704         LCONSOLE_ERROR_MSG(0x132, "%s: BAD WRITE CHECKSUM: %s: from %s inode "
1705                            DFID " object "DOSTID" extent [%llu-%llu], original "
1706                            "client csum %x (type %x), server csum %x (type %x),"
1707                            " client csum now %x\n",
1708                            obd_name, msg, libcfs_nid2str(peer->nid),
1709                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1710                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1711                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1712                            POSTID(&oa->o_oi), aa->aa_ppga[0]->off,
1713                            aa->aa_ppga[aa->aa_page_count - 1]->off +
1714                                 aa->aa_ppga[aa->aa_page_count-1]->count - 1,
1715                            client_cksum,
1716                            obd_cksum_type_unpack(aa->aa_oa->o_flags),
1717                            server_cksum, cksum_type, new_cksum);
1718         return 1;
1719 }
1720
1721 /* Note rc enters this function as number of bytes transferred */
1722 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1723 {
1724         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1725         struct client_obd *cli = aa->aa_cli;
1726         const char *obd_name = cli->cl_import->imp_obd->obd_name;
1727         const struct lnet_process_id *peer =
1728                 &req->rq_import->imp_connection->c_peer;
1729         struct ost_body *body;
1730         u32 client_cksum = 0;
1731         ENTRY;
1732
1733         if (rc < 0 && rc != -EDQUOT) {
1734                 DEBUG_REQ(D_INFO, req, "Failed request with rc = %d\n", rc);
1735                 RETURN(rc);
1736         }
1737
1738         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1739         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1740         if (body == NULL) {
1741                 DEBUG_REQ(D_INFO, req, "Can't unpack body\n");
1742                 RETURN(-EPROTO);
1743         }
1744
1745         /* set/clear over quota flag for a uid/gid/projid */
1746         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1747             body->oa.o_valid & (OBD_MD_FLALLQUOTA)) {
1748                 unsigned qid[LL_MAXQUOTAS] = {
1749                                          body->oa.o_uid, body->oa.o_gid,
1750                                          body->oa.o_projid };
1751                 CDEBUG(D_QUOTA, "setdq for [%u %u %u] with valid %#llx, flags %x\n",
1752                        body->oa.o_uid, body->oa.o_gid, body->oa.o_projid,
1753                        body->oa.o_valid, body->oa.o_flags);
1754                        osc_quota_setdq(cli, req->rq_xid, qid, body->oa.o_valid,
1755                                        body->oa.o_flags);
1756         }
1757
1758         osc_update_grant(cli, body);
1759
1760         if (rc < 0)
1761                 RETURN(rc);
1762
1763         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1764                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1765
1766         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1767                 if (rc > 0) {
1768                         CERROR("Unexpected +ve rc %d\n", rc);
1769                         RETURN(-EPROTO);
1770                 }
1771
1772                 if (req->rq_bulk != NULL &&
1773                     sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1774                         RETURN(-EAGAIN);
1775
1776                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1777                     check_write_checksum(&body->oa, peer, client_cksum,
1778                                          body->oa.o_cksum, aa))
1779                         RETURN(-EAGAIN);
1780
1781                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1782                                      aa->aa_page_count, aa->aa_ppga);
1783                 GOTO(out, rc);
1784         }
1785
1786         /* The rest of this function executes only for OST_READs */
1787
1788         if (req->rq_bulk == NULL) {
1789                 rc = req_capsule_get_size(&req->rq_pill, &RMF_SHORT_IO,
1790                                           RCL_SERVER);
1791                 LASSERT(rc == req->rq_status);
1792         } else {
1793                 /* if unwrap_bulk failed, return -EAGAIN to retry */
1794                 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1795         }
1796         if (rc < 0)
1797                 GOTO(out, rc = -EAGAIN);
1798
1799         if (rc > aa->aa_requested_nob) {
1800                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1801                        aa->aa_requested_nob);
1802                 RETURN(-EPROTO);
1803         }
1804
1805         if (req->rq_bulk != NULL && rc != req->rq_bulk->bd_nob_transferred) {
1806                 CERROR ("Unexpected rc %d (%d transferred)\n",
1807                         rc, req->rq_bulk->bd_nob_transferred);
1808                 return (-EPROTO);
1809         }
1810
1811         if (req->rq_bulk == NULL) {
1812                 /* short io */
1813                 int nob, pg_count, i = 0;
1814                 unsigned char *buf;
1815
1816                 CDEBUG(D_CACHE, "Using short io read, size %d\n", rc);
1817                 pg_count = aa->aa_page_count;
1818                 buf = req_capsule_server_sized_get(&req->rq_pill, &RMF_SHORT_IO,
1819                                                    rc);
1820                 nob = rc;
1821                 while (nob > 0 && pg_count > 0) {
1822                         unsigned char *ptr;
1823                         int count = aa->aa_ppga[i]->count > nob ?
1824                                     nob : aa->aa_ppga[i]->count;
1825
1826                         CDEBUG(D_CACHE, "page %p count %d\n",
1827                                aa->aa_ppga[i]->pg, count);
1828                         ptr = ll_kmap_atomic(aa->aa_ppga[i]->pg, KM_USER0);
1829                         memcpy(ptr + (aa->aa_ppga[i]->off & ~PAGE_MASK), buf,
1830                                count);
1831                         ll_kunmap_atomic((void *) ptr, KM_USER0);
1832
1833                         buf += count;
1834                         nob -= count;
1835                         i++;
1836                         pg_count--;
1837                 }
1838         }
1839
1840         if (rc < aa->aa_requested_nob)
1841                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1842
1843         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1844                 static int cksum_counter;
1845                 u32        server_cksum = body->oa.o_cksum;
1846                 char      *via = "";
1847                 char      *router = "";
1848                 enum cksum_types cksum_type;
1849                 u32 o_flags = body->oa.o_valid & OBD_MD_FLFLAGS ?
1850                         body->oa.o_flags : 0;
1851
1852                 cksum_type = obd_cksum_type_unpack(o_flags);
1853                 rc = osc_checksum_bulk_rw(obd_name, cksum_type, rc,
1854                                           aa->aa_page_count, aa->aa_ppga,
1855                                           OST_READ, &client_cksum);
1856                 if (rc < 0)
1857                         GOTO(out, rc);
1858
1859                 if (req->rq_bulk != NULL &&
1860                     peer->nid != req->rq_bulk->bd_sender) {
1861                         via = " via ";
1862                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1863                 }
1864
1865                 if (server_cksum != client_cksum) {
1866                         struct ost_body *clbody;
1867                         u32 page_count = aa->aa_page_count;
1868
1869                         clbody = req_capsule_client_get(&req->rq_pill,
1870                                                         &RMF_OST_BODY);
1871                         if (cli->cl_checksum_dump)
1872                                 dump_all_bulk_pages(&clbody->oa, page_count,
1873                                                     aa->aa_ppga, server_cksum,
1874                                                     client_cksum);
1875
1876                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1877                                            "%s%s%s inode "DFID" object "DOSTID
1878                                            " extent [%llu-%llu], client %x, "
1879                                            "server %x, cksum_type %x\n",
1880                                            obd_name,
1881                                            libcfs_nid2str(peer->nid),
1882                                            via, router,
1883                                            clbody->oa.o_valid & OBD_MD_FLFID ?
1884                                                 clbody->oa.o_parent_seq : 0ULL,
1885                                            clbody->oa.o_valid & OBD_MD_FLFID ?
1886                                                 clbody->oa.o_parent_oid : 0,
1887                                            clbody->oa.o_valid & OBD_MD_FLFID ?
1888                                                 clbody->oa.o_parent_ver : 0,
1889                                            POSTID(&body->oa.o_oi),
1890                                            aa->aa_ppga[0]->off,
1891                                            aa->aa_ppga[page_count-1]->off +
1892                                            aa->aa_ppga[page_count-1]->count - 1,
1893                                            client_cksum, server_cksum,
1894                                            cksum_type);
1895                         cksum_counter = 0;
1896                         aa->aa_oa->o_cksum = client_cksum;
1897                         rc = -EAGAIN;
1898                 } else {
1899                         cksum_counter++;
1900                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1901                         rc = 0;
1902                 }
1903         } else if (unlikely(client_cksum)) {
1904                 static int cksum_missed;
1905
1906                 cksum_missed++;
1907                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1908                         CERROR("Checksum %u requested from %s but not sent\n",
1909                                cksum_missed, libcfs_nid2str(peer->nid));
1910         } else {
1911                 rc = 0;
1912         }
1913 out:
1914         if (rc >= 0)
1915                 lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
1916                                      aa->aa_oa, &body->oa);
1917
1918         RETURN(rc);
1919 }
1920
1921 static int osc_brw_redo_request(struct ptlrpc_request *request,
1922                                 struct osc_brw_async_args *aa, int rc)
1923 {
1924         struct ptlrpc_request *new_req;
1925         struct osc_brw_async_args *new_aa;
1926         struct osc_async_page *oap;
1927         ENTRY;
1928
1929         DEBUG_REQ(rc == -EINPROGRESS ? D_RPCTRACE : D_ERROR, request,
1930                   "redo for recoverable error %d", rc);
1931
1932         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1933                                 OST_WRITE ? OBD_BRW_WRITE : OBD_BRW_READ,
1934                                   aa->aa_cli, aa->aa_oa, aa->aa_page_count,
1935                                   aa->aa_ppga, &new_req, 1);
1936         if (rc)
1937                 RETURN(rc);
1938
1939         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1940                 if (oap->oap_request != NULL) {
1941                         LASSERTF(request == oap->oap_request,
1942                                  "request %p != oap_request %p\n",
1943                                  request, oap->oap_request);
1944                         if (oap->oap_interrupted) {
1945                                 ptlrpc_req_finished(new_req);
1946                                 RETURN(-EINTR);
1947                         }
1948                 }
1949         }
1950         /*
1951          * New request takes over pga and oaps from old request.
1952          * Note that copying a list_head doesn't work, need to move it...
1953          */
1954         aa->aa_resends++;
1955         new_req->rq_interpret_reply = request->rq_interpret_reply;
1956         new_req->rq_async_args = request->rq_async_args;
1957         new_req->rq_commit_cb = request->rq_commit_cb;
1958         /* cap resend delay to the current request timeout, this is similar to
1959          * what ptlrpc does (see after_reply()) */
1960         if (aa->aa_resends > new_req->rq_timeout)
1961                 new_req->rq_sent = ktime_get_real_seconds() + new_req->rq_timeout;
1962         else
1963                 new_req->rq_sent = ktime_get_real_seconds() + aa->aa_resends;
1964         new_req->rq_generation_set = 1;
1965         new_req->rq_import_generation = request->rq_import_generation;
1966
1967         new_aa = ptlrpc_req_async_args(new_req);
1968
1969         INIT_LIST_HEAD(&new_aa->aa_oaps);
1970         list_splice_init(&aa->aa_oaps, &new_aa->aa_oaps);
1971         INIT_LIST_HEAD(&new_aa->aa_exts);
1972         list_splice_init(&aa->aa_exts, &new_aa->aa_exts);
1973         new_aa->aa_resends = aa->aa_resends;
1974
1975         list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1976                 if (oap->oap_request) {
1977                         ptlrpc_req_finished(oap->oap_request);
1978                         oap->oap_request = ptlrpc_request_addref(new_req);
1979                 }
1980         }
1981
1982         /* XXX: This code will run into problem if we're going to support
1983          * to add a series of BRW RPCs into a self-defined ptlrpc_request_set
1984          * and wait for all of them to be finished. We should inherit request
1985          * set from old request. */
1986         ptlrpcd_add_req(new_req);
1987
1988         DEBUG_REQ(D_INFO, new_req, "new request");
1989         RETURN(0);
1990 }
1991
1992 /*
1993  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1994  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1995  * fine for our small page arrays and doesn't require allocation.  its an
1996  * insertion sort that swaps elements that are strides apart, shrinking the
1997  * stride down until its '1' and the array is sorted.
1998  */
1999 static void sort_brw_pages(struct brw_page **array, int num)
2000 {
2001         int stride, i, j;
2002         struct brw_page *tmp;
2003
2004         if (num == 1)
2005                 return;
2006         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
2007                 ;
2008
2009         do {
2010                 stride /= 3;
2011                 for (i = stride ; i < num ; i++) {
2012                         tmp = array[i];
2013                         j = i;
2014                         while (j >= stride && array[j - stride]->off > tmp->off) {
2015                                 array[j] = array[j - stride];
2016                                 j -= stride;
2017                         }
2018                         array[j] = tmp;
2019                 }
2020         } while (stride > 1);
2021 }
2022
2023 static void osc_release_ppga(struct brw_page **ppga, size_t count)
2024 {
2025         LASSERT(ppga != NULL);
2026         OBD_FREE(ppga, sizeof(*ppga) * count);
2027 }
2028
2029 static int brw_interpret(const struct lu_env *env,
2030                          struct ptlrpc_request *req, void *args, int rc)
2031 {
2032         struct osc_brw_async_args *aa = args;
2033         struct osc_extent *ext;
2034         struct osc_extent *tmp;
2035         struct client_obd *cli = aa->aa_cli;
2036         unsigned long transferred = 0;
2037
2038         ENTRY;
2039
2040         rc = osc_brw_fini_request(req, rc);
2041         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2042         /*
2043          * When server returns -EINPROGRESS, client should always retry
2044          * regardless of the number of times the bulk was resent already.
2045          */
2046         if (osc_recoverable_error(rc) && !req->rq_no_delay) {
2047                 if (req->rq_import_generation !=
2048                     req->rq_import->imp_generation) {
2049                         CDEBUG(D_HA, "%s: resend cross eviction for object: "
2050                                ""DOSTID", rc = %d.\n",
2051                                req->rq_import->imp_obd->obd_name,
2052                                POSTID(&aa->aa_oa->o_oi), rc);
2053                 } else if (rc == -EINPROGRESS ||
2054                     client_should_resend(aa->aa_resends, aa->aa_cli)) {
2055                         rc = osc_brw_redo_request(req, aa, rc);
2056                 } else {
2057                         CERROR("%s: too many resent retries for object: "
2058                                "%llu:%llu, rc = %d.\n",
2059                                req->rq_import->imp_obd->obd_name,
2060                                POSTID(&aa->aa_oa->o_oi), rc);
2061                 }
2062
2063                 if (rc == 0)
2064                         RETURN(0);
2065                 else if (rc == -EAGAIN || rc == -EINPROGRESS)
2066                         rc = -EIO;
2067         }
2068
2069         if (rc == 0) {
2070                 struct obdo *oa = aa->aa_oa;
2071                 struct cl_attr *attr = &osc_env_info(env)->oti_attr;
2072                 unsigned long valid = 0;
2073                 struct cl_object *obj;
2074                 struct osc_async_page *last;
2075
2076                 last = brw_page2oap(aa->aa_ppga[aa->aa_page_count - 1]);
2077                 obj = osc2cl(last->oap_obj);
2078
2079                 cl_object_attr_lock(obj);
2080                 if (oa->o_valid & OBD_MD_FLBLOCKS) {
2081                         attr->cat_blocks = oa->o_blocks;
2082                         valid |= CAT_BLOCKS;
2083                 }
2084                 if (oa->o_valid & OBD_MD_FLMTIME) {
2085                         attr->cat_mtime = oa->o_mtime;
2086                         valid |= CAT_MTIME;
2087                 }
2088                 if (oa->o_valid & OBD_MD_FLATIME) {
2089                         attr->cat_atime = oa->o_atime;
2090                         valid |= CAT_ATIME;
2091                 }
2092                 if (oa->o_valid & OBD_MD_FLCTIME) {
2093                         attr->cat_ctime = oa->o_ctime;
2094                         valid |= CAT_CTIME;
2095                 }
2096
2097                 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
2098                         struct lov_oinfo *loi = cl2osc(obj)->oo_oinfo;
2099                         loff_t last_off = last->oap_count + last->oap_obj_off +
2100                                 last->oap_page_off;
2101
2102                         /* Change file size if this is an out of quota or
2103                          * direct IO write and it extends the file size */
2104                         if (loi->loi_lvb.lvb_size < last_off) {
2105                                 attr->cat_size = last_off;
2106                                 valid |= CAT_SIZE;
2107                         }
2108                         /* Extend KMS if it's not a lockless write */
2109                         if (loi->loi_kms < last_off &&
2110                             oap2osc_page(last)->ops_srvlock == 0) {
2111                                 attr->cat_kms = last_off;
2112                                 valid |= CAT_KMS;
2113                         }
2114                 }
2115
2116                 if (valid != 0)
2117                         cl_object_attr_update(env, obj, attr, valid);
2118                 cl_object_attr_unlock(obj);
2119         }
2120         OBD_SLAB_FREE_PTR(aa->aa_oa, osc_obdo_kmem);
2121
2122         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE && rc == 0)
2123                 osc_inc_unstable_pages(req);
2124
2125         list_for_each_entry_safe(ext, tmp, &aa->aa_exts, oe_link) {
2126                 list_del_init(&ext->oe_link);
2127                 osc_extent_finish(env, ext, 1,
2128                                   rc && req->rq_no_delay ? -EWOULDBLOCK : rc);
2129         }
2130         LASSERT(list_empty(&aa->aa_exts));
2131         LASSERT(list_empty(&aa->aa_oaps));
2132
2133         transferred = (req->rq_bulk == NULL ? /* short io */
2134                        aa->aa_requested_nob :
2135                        req->rq_bulk->bd_nob_transferred);
2136
2137         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2138         ptlrpc_lprocfs_brw(req, transferred);
2139
2140         spin_lock(&cli->cl_loi_list_lock);
2141         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2142          * is called so we know whether to go to sync BRWs or wait for more
2143          * RPCs to complete */
2144         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2145                 cli->cl_w_in_flight--;
2146         else
2147                 cli->cl_r_in_flight--;
2148         osc_wake_cache_waiters(cli);
2149         spin_unlock(&cli->cl_loi_list_lock);
2150
2151         osc_io_unplug(env, cli, NULL);
2152         RETURN(rc);
2153 }
2154
2155 static void brw_commit(struct ptlrpc_request *req)
2156 {
2157         /* If osc_inc_unstable_pages (via osc_extent_finish) races with
2158          * this called via the rq_commit_cb, I need to ensure
2159          * osc_dec_unstable_pages is still called. Otherwise unstable
2160          * pages may be leaked. */
2161         spin_lock(&req->rq_lock);
2162         if (likely(req->rq_unstable)) {
2163                 req->rq_unstable = 0;
2164                 spin_unlock(&req->rq_lock);
2165
2166                 osc_dec_unstable_pages(req);
2167         } else {
2168                 req->rq_committed = 1;
2169                 spin_unlock(&req->rq_lock);
2170         }
2171 }
2172
2173 /**
2174  * Build an RPC by the list of extent @ext_list. The caller must ensure
2175  * that the total pages in this list are NOT over max pages per RPC.
2176  * Extents in the list must be in OES_RPC state.
2177  */
2178 int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
2179                   struct list_head *ext_list, int cmd)
2180 {
2181         struct ptlrpc_request           *req = NULL;
2182         struct osc_extent               *ext;
2183         struct brw_page                 **pga = NULL;
2184         struct osc_brw_async_args       *aa = NULL;
2185         struct obdo                     *oa = NULL;
2186         struct osc_async_page           *oap;
2187         struct osc_object               *obj = NULL;
2188         struct cl_req_attr              *crattr = NULL;
2189         loff_t                          starting_offset = OBD_OBJECT_EOF;
2190         loff_t                          ending_offset = 0;
2191         int                             mpflag = 0;
2192         int                             mem_tight = 0;
2193         int                             page_count = 0;
2194         bool                            soft_sync = false;
2195         bool                            interrupted = false;
2196         bool                            ndelay = false;
2197         int                             i;
2198         int                             grant = 0;
2199         int                             rc;
2200         __u32                           layout_version = 0;
2201         struct list_head                rpc_list = LIST_HEAD_INIT(rpc_list);
2202         struct ost_body                 *body;
2203         ENTRY;
2204         LASSERT(!list_empty(ext_list));
2205
2206         /* add pages into rpc_list to build BRW rpc */
2207         list_for_each_entry(ext, ext_list, oe_link) {
2208                 LASSERT(ext->oe_state == OES_RPC);
2209                 mem_tight |= ext->oe_memalloc;
2210                 grant += ext->oe_grants;
2211                 page_count += ext->oe_nr_pages;
2212                 layout_version = MAX(layout_version, ext->oe_layout_version);
2213                 if (obj == NULL)
2214                         obj = ext->oe_obj;
2215         }
2216
2217         soft_sync = osc_over_unstable_soft_limit(cli);
2218         if (mem_tight)
2219                 mpflag = cfs_memory_pressure_get_and_set();
2220
2221         OBD_ALLOC(pga, sizeof(*pga) * page_count);
2222         if (pga == NULL)
2223                 GOTO(out, rc = -ENOMEM);
2224
2225         OBD_SLAB_ALLOC_PTR_GFP(oa, osc_obdo_kmem, GFP_NOFS);
2226         if (oa == NULL)
2227                 GOTO(out, rc = -ENOMEM);
2228
2229         i = 0;
2230         list_for_each_entry(ext, ext_list, oe_link) {
2231                 list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
2232                         if (mem_tight)
2233                                 oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
2234                         if (soft_sync)
2235                                 oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
2236                         pga[i] = &oap->oap_brw_page;
2237                         pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2238                         i++;
2239
2240                         list_add_tail(&oap->oap_rpc_item, &rpc_list);
2241                         if (starting_offset == OBD_OBJECT_EOF ||
2242                             starting_offset > oap->oap_obj_off)
2243                                 starting_offset = oap->oap_obj_off;
2244                         else
2245                                 LASSERT(oap->oap_page_off == 0);
2246                         if (ending_offset < oap->oap_obj_off + oap->oap_count)
2247                                 ending_offset = oap->oap_obj_off +
2248                                                 oap->oap_count;
2249                         else
2250                                 LASSERT(oap->oap_page_off + oap->oap_count ==
2251                                         PAGE_SIZE);
2252                         if (oap->oap_interrupted)
2253                                 interrupted = true;
2254                 }
2255                 if (ext->oe_ndelay)
2256                         ndelay = true;
2257         }
2258
2259         /* first page in the list */
2260         oap = list_entry(rpc_list.next, typeof(*oap), oap_rpc_item);
2261
2262         crattr = &osc_env_info(env)->oti_req_attr;
2263         memset(crattr, 0, sizeof(*crattr));
2264         crattr->cra_type = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2265         crattr->cra_flags = ~0ULL;
2266         crattr->cra_page = oap2cl_page(oap);
2267         crattr->cra_oa = oa;
2268         cl_req_attr_set(env, osc2cl(obj), crattr);
2269
2270         if (cmd == OBD_BRW_WRITE) {
2271                 oa->o_grant_used = grant;
2272                 if (layout_version > 0) {
2273                         CDEBUG(D_LAYOUT, DFID": write with layout version %u\n",
2274                                PFID(&oa->o_oi.oi_fid), layout_version);
2275
2276                         oa->o_layout_version = layout_version;
2277                         oa->o_valid |= OBD_MD_LAYOUT_VERSION;
2278                 }
2279         }
2280
2281         sort_brw_pages(pga, page_count);
2282         rc = osc_brw_prep_request(cmd, cli, oa, page_count, pga, &req, 0);
2283         if (rc != 0) {
2284                 CERROR("prep_req failed: %d\n", rc);
2285                 GOTO(out, rc);
2286         }
2287
2288         req->rq_commit_cb = brw_commit;
2289         req->rq_interpret_reply = brw_interpret;
2290         req->rq_memalloc = mem_tight != 0;
2291         oap->oap_request = ptlrpc_request_addref(req);
2292         if (interrupted && !req->rq_intr)
2293                 ptlrpc_mark_interrupted(req);
2294         if (ndelay) {
2295                 req->rq_no_resend = req->rq_no_delay = 1;
2296                 /* probably set a shorter timeout value.
2297                  * to handle ETIMEDOUT in brw_interpret() correctly. */
2298                 /* lustre_msg_set_timeout(req, req->rq_timeout / 2); */
2299         }
2300
2301         /* Need to update the timestamps after the request is built in case
2302          * we race with setattr (locally or in queue at OST).  If OST gets
2303          * later setattr before earlier BRW (as determined by the request xid),
2304          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2305          * way to do this in a single call.  bug 10150 */
2306         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2307         crattr->cra_oa = &body->oa;
2308         crattr->cra_flags = OBD_MD_FLMTIME | OBD_MD_FLCTIME | OBD_MD_FLATIME;
2309         cl_req_attr_set(env, osc2cl(obj), crattr);
2310         lustre_msg_set_jobid(req->rq_reqmsg, crattr->cra_jobid);
2311
2312         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2313         aa = ptlrpc_req_async_args(req);
2314         INIT_LIST_HEAD(&aa->aa_oaps);
2315         list_splice_init(&rpc_list, &aa->aa_oaps);
2316         INIT_LIST_HEAD(&aa->aa_exts);
2317         list_splice_init(ext_list, &aa->aa_exts);
2318
2319         spin_lock(&cli->cl_loi_list_lock);
2320         starting_offset >>= PAGE_SHIFT;
2321         if (cmd == OBD_BRW_READ) {
2322                 cli->cl_r_in_flight++;
2323                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2324                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2325                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2326                                       starting_offset + 1);
2327         } else {
2328                 cli->cl_w_in_flight++;
2329                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2330                 lprocfs_oh_tally(&cli->cl_write_rpc_hist, cli->cl_w_in_flight);
2331                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2332                                       starting_offset + 1);
2333         }
2334         spin_unlock(&cli->cl_loi_list_lock);
2335
2336         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %ur/%uw in flight",
2337                   page_count, aa, cli->cl_r_in_flight,
2338                   cli->cl_w_in_flight);
2339         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_DELAY_IO, cfs_fail_val);
2340
2341         ptlrpcd_add_req(req);
2342         rc = 0;
2343         EXIT;
2344
2345 out:
2346         if (mem_tight != 0)
2347                 cfs_memory_pressure_restore(mpflag);
2348
2349         if (rc != 0) {
2350                 LASSERT(req == NULL);
2351
2352                 if (oa)
2353                         OBD_SLAB_FREE_PTR(oa, osc_obdo_kmem);
2354                 if (pga)
2355                         OBD_FREE(pga, sizeof(*pga) * page_count);
2356                 /* this should happen rarely and is pretty bad, it makes the
2357                  * pending list not follow the dirty order */
2358                 while (!list_empty(ext_list)) {
2359                         ext = list_entry(ext_list->next, struct osc_extent,
2360                                          oe_link);
2361                         list_del_init(&ext->oe_link);
2362                         osc_extent_finish(env, ext, 0, rc);
2363                 }
2364         }
2365         RETURN(rc);
2366 }
2367
2368 static int osc_set_lock_data(struct ldlm_lock *lock, void *data)
2369 {
2370         int set = 0;
2371
2372         LASSERT(lock != NULL);
2373
2374         lock_res_and_lock(lock);
2375
2376         if (lock->l_ast_data == NULL)
2377                 lock->l_ast_data = data;
2378         if (lock->l_ast_data == data)
2379                 set = 1;
2380
2381         unlock_res_and_lock(lock);
2382
2383         return set;
2384 }
2385
2386 int osc_enqueue_fini(struct ptlrpc_request *req, osc_enqueue_upcall_f upcall,
2387                      void *cookie, struct lustre_handle *lockh,
2388                      enum ldlm_mode mode, __u64 *flags, bool speculative,
2389                      int errcode)
2390 {
2391         bool intent = *flags & LDLM_FL_HAS_INTENT;
2392         int rc;
2393         ENTRY;
2394
2395         /* The request was created before ldlm_cli_enqueue call. */
2396         if (intent && errcode == ELDLM_LOCK_ABORTED) {
2397                 struct ldlm_reply *rep;
2398
2399                 rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
2400                 LASSERT(rep != NULL);
2401
2402                 rep->lock_policy_res1 =
2403                         ptlrpc_status_ntoh(rep->lock_policy_res1);
2404                 if (rep->lock_policy_res1)
2405                         errcode = rep->lock_policy_res1;
2406                 if (!speculative)
2407                         *flags |= LDLM_FL_LVB_READY;
2408         } else if (errcode == ELDLM_OK) {
2409                 *flags |= LDLM_FL_LVB_READY;
2410         }
2411
2412         /* Call the update callback. */
2413         rc = (*upcall)(cookie, lockh, errcode);
2414
2415         /* release the reference taken in ldlm_cli_enqueue() */
2416         if (errcode == ELDLM_LOCK_MATCHED)
2417                 errcode = ELDLM_OK;
2418         if (errcode == ELDLM_OK && lustre_handle_is_used(lockh))
2419                 ldlm_lock_decref(lockh, mode);
2420
2421         RETURN(rc);
2422 }
2423
2424 int osc_enqueue_interpret(const struct lu_env *env, struct ptlrpc_request *req,
2425                           void *args, int rc)
2426 {
2427         struct osc_enqueue_args *aa = args;
2428         struct ldlm_lock *lock;
2429         struct lustre_handle *lockh = &aa->oa_lockh;
2430         enum ldlm_mode mode = aa->oa_mode;
2431         struct ost_lvb *lvb = aa->oa_lvb;
2432         __u32 lvb_len = sizeof(*lvb);
2433         __u64 flags = 0;
2434
2435         ENTRY;
2436
2437         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
2438          * be valid. */
2439         lock = ldlm_handle2lock(lockh);
2440         LASSERTF(lock != NULL,
2441                  "lockh %#llx, req %p, aa %p - client evicted?\n",
2442                  lockh->cookie, req, aa);
2443
2444         /* Take an additional reference so that a blocking AST that
2445          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
2446          * to arrive after an upcall has been executed by
2447          * osc_enqueue_fini(). */
2448         ldlm_lock_addref(lockh, mode);
2449
2450         /* Let cl_lock_state_wait fail with -ERESTARTSYS to unuse sublocks. */
2451         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_HANG, 2);
2452
2453         /* Let CP AST to grant the lock first. */
2454         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
2455
2456         if (aa->oa_speculative) {
2457                 LASSERT(aa->oa_lvb == NULL);
2458                 LASSERT(aa->oa_flags == NULL);
2459                 aa->oa_flags = &flags;
2460         }
2461
2462         /* Complete obtaining the lock procedure. */
2463         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_type, 1,
2464                                    aa->oa_mode, aa->oa_flags, lvb, lvb_len,
2465                                    lockh, rc);
2466         /* Complete osc stuff. */
2467         rc = osc_enqueue_fini(req, aa->oa_upcall, aa->oa_cookie, lockh, mode,
2468                               aa->oa_flags, aa->oa_speculative, rc);
2469
2470         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
2471
2472         ldlm_lock_decref(lockh, mode);
2473         LDLM_LOCK_PUT(lock);
2474         RETURN(rc);
2475 }
2476
2477 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
2478
2479 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
2480  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
2481  * other synchronous requests, however keeping some locks and trying to obtain
2482  * others may take a considerable amount of time in a case of ost failure; and
2483  * when other sync requests do not get released lock from a client, the client
2484  * is evicted from the cluster -- such scenarious make the life difficult, so
2485  * release locks just after they are obtained. */
2486 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2487                      __u64 *flags, union ldlm_policy_data *policy,
2488                      struct ost_lvb *lvb, int kms_valid,
2489                      osc_enqueue_upcall_f upcall, void *cookie,
2490                      struct ldlm_enqueue_info *einfo,
2491                      struct ptlrpc_request_set *rqset, int async,
2492                      bool speculative)
2493 {
2494         struct obd_device *obd = exp->exp_obd;
2495         struct lustre_handle lockh = { 0 };
2496         struct ptlrpc_request *req = NULL;
2497         int intent = *flags & LDLM_FL_HAS_INTENT;
2498         __u64 match_flags = *flags;
2499         enum ldlm_mode mode;
2500         int rc;
2501         ENTRY;
2502
2503         /* Filesystem lock extents are extended to page boundaries so that
2504          * dealing with the page cache is a little smoother.  */
2505         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2506         policy->l_extent.end |= ~PAGE_MASK;
2507
2508         /*
2509          * kms is not valid when either object is completely fresh (so that no
2510          * locks are cached), or object was evicted. In the latter case cached
2511          * lock cannot be used, because it would prime inode state with
2512          * potentially stale LVB.
2513          */
2514         if (!kms_valid)
2515                 goto no_match;
2516
2517         /* Next, search for already existing extent locks that will cover us */
2518         /* If we're trying to read, we also search for an existing PW lock.  The
2519          * VFS and page cache already protect us locally, so lots of readers/
2520          * writers can share a single PW lock.
2521          *
2522          * There are problems with conversion deadlocks, so instead of
2523          * converting a read lock to a write lock, we'll just enqueue a new
2524          * one.
2525          *
2526          * At some point we should cancel the read lock instead of making them
2527          * send us a blocking callback, but there are problems with canceling
2528          * locks out from other users right now, too. */
2529         mode = einfo->ei_mode;
2530         if (einfo->ei_mode == LCK_PR)
2531                 mode |= LCK_PW;
2532         /* Normal lock requests must wait for the LVB to be ready before
2533          * matching a lock; speculative lock requests do not need to,
2534          * because they will not actually use the lock. */
2535         if (!speculative)
2536                 match_flags |= LDLM_FL_LVB_READY;
2537         if (intent != 0)
2538                 match_flags |= LDLM_FL_BLOCK_GRANTED;
2539         mode = ldlm_lock_match(obd->obd_namespace, match_flags, res_id,
2540                                einfo->ei_type, policy, mode, &lockh, 0);
2541         if (mode) {
2542                 struct ldlm_lock *matched;
2543
2544                 if (*flags & LDLM_FL_TEST_LOCK)
2545                         RETURN(ELDLM_OK);
2546
2547                 matched = ldlm_handle2lock(&lockh);
2548                 if (speculative) {
2549                         /* This DLM lock request is speculative, and does not
2550                          * have an associated IO request. Therefore if there
2551                          * is already a DLM lock, it wll just inform the
2552                          * caller to cancel the request for this stripe.*/
2553                         lock_res_and_lock(matched);
2554                         if (ldlm_extent_equal(&policy->l_extent,
2555                             &matched->l_policy_data.l_extent))
2556                                 rc = -EEXIST;
2557                         else
2558                                 rc = -ECANCELED;
2559                         unlock_res_and_lock(matched);
2560
2561                         ldlm_lock_decref(&lockh, mode);
2562                         LDLM_LOCK_PUT(matched);
2563                         RETURN(rc);
2564                 } else if (osc_set_lock_data(matched, einfo->ei_cbdata)) {
2565                         *flags |= LDLM_FL_LVB_READY;
2566
2567                         /* We already have a lock, and it's referenced. */
2568                         (*upcall)(cookie, &lockh, ELDLM_LOCK_MATCHED);
2569
2570                         ldlm_lock_decref(&lockh, mode);
2571                         LDLM_LOCK_PUT(matched);
2572                         RETURN(ELDLM_OK);
2573                 } else {
2574                         ldlm_lock_decref(&lockh, mode);
2575                         LDLM_LOCK_PUT(matched);
2576                 }
2577         }
2578
2579 no_match:
2580         if (*flags & (LDLM_FL_TEST_LOCK | LDLM_FL_MATCH_LOCK))
2581                 RETURN(-ENOLCK);
2582
2583         if (intent) {
2584                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2585                                            &RQF_LDLM_ENQUEUE_LVB);
2586                 if (req == NULL)
2587                         RETURN(-ENOMEM);
2588
2589                 rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
2590                 if (rc) {
2591                         ptlrpc_request_free(req);
2592                         RETURN(rc);
2593                 }
2594
2595                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
2596                                      sizeof *lvb);
2597                 ptlrpc_request_set_replen(req);
2598         }
2599
2600         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
2601         *flags &= ~LDLM_FL_BLOCK_GRANTED;
2602
2603         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
2604                               sizeof(*lvb), LVB_T_OST, &lockh, async);
2605         if (async) {
2606                 if (!rc) {
2607                         struct osc_enqueue_args *aa;
2608                         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2609                         aa = ptlrpc_req_async_args(req);
2610                         aa->oa_exp         = exp;
2611                         aa->oa_mode        = einfo->ei_mode;
2612                         aa->oa_type        = einfo->ei_type;
2613                         lustre_handle_copy(&aa->oa_lockh, &lockh);
2614                         aa->oa_upcall      = upcall;
2615                         aa->oa_cookie      = cookie;
2616                         aa->oa_speculative = speculative;
2617                         if (!speculative) {
2618                                 aa->oa_flags  = flags;
2619                                 aa->oa_lvb    = lvb;
2620                         } else {
2621                                 /* speculative locks are essentially to enqueue
2622                                  * a DLM lock  in advance, so we don't care
2623                                  * about the result of the enqueue. */
2624                                 aa->oa_lvb    = NULL;
2625                                 aa->oa_flags  = NULL;
2626                         }
2627
2628                         req->rq_interpret_reply = osc_enqueue_interpret;
2629                         if (rqset == PTLRPCD_SET)
2630                                 ptlrpcd_add_req(req);
2631                         else
2632                                 ptlrpc_set_add_req(rqset, req);
2633                 } else if (intent) {
2634                         ptlrpc_req_finished(req);
2635                 }
2636                 RETURN(rc);
2637         }
2638
2639         rc = osc_enqueue_fini(req, upcall, cookie, &lockh, einfo->ei_mode,
2640                               flags, speculative, rc);
2641         if (intent)
2642                 ptlrpc_req_finished(req);
2643
2644         RETURN(rc);
2645 }
2646
2647 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2648                    enum ldlm_type type, union ldlm_policy_data *policy,
2649                    enum ldlm_mode mode, __u64 *flags, void *data,
2650                    struct lustre_handle *lockh, int unref)
2651 {
2652         struct obd_device *obd = exp->exp_obd;
2653         __u64 lflags = *flags;
2654         enum ldlm_mode rc;
2655         ENTRY;
2656
2657         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
2658                 RETURN(-EIO);
2659
2660         /* Filesystem lock extents are extended to page boundaries so that
2661          * dealing with the page cache is a little smoother */
2662         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2663         policy->l_extent.end |= ~PAGE_MASK;
2664
2665         /* Next, search for already existing extent locks that will cover us */
2666         /* If we're trying to read, we also search for an existing PW lock.  The
2667          * VFS and page cache already protect us locally, so lots of readers/
2668          * writers can share a single PW lock. */
2669         rc = mode;
2670         if (mode == LCK_PR)
2671                 rc |= LCK_PW;
2672         rc = ldlm_lock_match(obd->obd_namespace, lflags,
2673                              res_id, type, policy, rc, lockh, unref);
2674         if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
2675                 RETURN(rc);
2676
2677         if (data != NULL) {
2678                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2679
2680                 LASSERT(lock != NULL);
2681                 if (!osc_set_lock_data(lock, data)) {
2682                         ldlm_lock_decref(lockh, rc);
2683                         rc = 0;
2684                 }
2685                 LDLM_LOCK_PUT(lock);
2686         }
2687         RETURN(rc);
2688 }
2689
2690 static int osc_statfs_interpret(const struct lu_env *env,
2691                                 struct ptlrpc_request *req, void *args, int rc)
2692 {
2693         struct osc_async_args *aa = args;
2694         struct obd_statfs *msfs;
2695
2696         ENTRY;
2697         if (rc == -EBADR)
2698                 /*
2699                  * The request has in fact never been sent due to issues at
2700                  * a higher level (LOV).  Exit immediately since the caller
2701                  * is aware of the problem and takes care of the clean up.
2702                  */
2703                 RETURN(rc);
2704
2705         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
2706             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
2707                 GOTO(out, rc = 0);
2708
2709         if (rc != 0)
2710                 GOTO(out, rc);
2711
2712         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2713         if (msfs == NULL)
2714                 GOTO(out, rc = -EPROTO);
2715
2716         *aa->aa_oi->oi_osfs = *msfs;
2717 out:
2718         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
2719
2720         RETURN(rc);
2721 }
2722
2723 static int osc_statfs_async(struct obd_export *exp,
2724                             struct obd_info *oinfo, time64_t max_age,
2725                             struct ptlrpc_request_set *rqset)
2726 {
2727         struct obd_device     *obd = class_exp2obd(exp);
2728         struct ptlrpc_request *req;
2729         struct osc_async_args *aa;
2730         int rc;
2731         ENTRY;
2732
2733         if (obd->obd_osfs_age >= max_age) {
2734                 CDEBUG(D_SUPER,
2735                        "%s: use %p cache blocks %llu/%llu objects %llu/%llu\n",
2736                        obd->obd_name, &obd->obd_osfs,
2737                        obd->obd_osfs.os_bavail, obd->obd_osfs.os_blocks,
2738                        obd->obd_osfs.os_ffree, obd->obd_osfs.os_files);
2739                 spin_lock(&obd->obd_osfs_lock);
2740                 memcpy(oinfo->oi_osfs, &obd->obd_osfs, sizeof(*oinfo->oi_osfs));
2741                 spin_unlock(&obd->obd_osfs_lock);
2742                 oinfo->oi_flags |= OBD_STATFS_FROM_CACHE;
2743                 if (oinfo->oi_cb_up)
2744                         oinfo->oi_cb_up(oinfo, 0);
2745
2746                 RETURN(0);
2747         }
2748
2749         /* We could possibly pass max_age in the request (as an absolute
2750          * timestamp or a "seconds.usec ago") so the target can avoid doing
2751          * extra calls into the filesystem if that isn't necessary (e.g.
2752          * during mount that would help a bit).  Having relative timestamps
2753          * is not so great if request processing is slow, while absolute
2754          * timestamps are not ideal because they need time synchronization. */
2755         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
2756         if (req == NULL)
2757                 RETURN(-ENOMEM);
2758
2759         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2760         if (rc) {
2761                 ptlrpc_request_free(req);
2762                 RETURN(rc);
2763         }
2764         ptlrpc_request_set_replen(req);
2765         req->rq_request_portal = OST_CREATE_PORTAL;
2766         ptlrpc_at_set_req_timeout(req);
2767
2768         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
2769                 /* procfs requests not want stat in wait for avoid deadlock */
2770                 req->rq_no_resend = 1;
2771                 req->rq_no_delay = 1;
2772         }
2773
2774         req->rq_interpret_reply = osc_statfs_interpret;
2775         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2776         aa = ptlrpc_req_async_args(req);
2777         aa->aa_oi = oinfo;
2778
2779         ptlrpc_set_add_req(rqset, req);
2780         RETURN(0);
2781 }
2782
2783 static int osc_statfs(const struct lu_env *env, struct obd_export *exp,
2784                       struct obd_statfs *osfs, time64_t max_age, __u32 flags)
2785 {
2786         struct obd_device     *obd = class_exp2obd(exp);
2787         struct obd_statfs     *msfs;
2788         struct ptlrpc_request *req;
2789         struct obd_import     *imp = NULL;
2790         int rc;
2791         ENTRY;
2792
2793
2794         /*Since the request might also come from lprocfs, so we need
2795          *sync this with client_disconnect_export Bug15684*/
2796         down_read(&obd->u.cli.cl_sem);
2797         if (obd->u.cli.cl_import)
2798                 imp = class_import_get(obd->u.cli.cl_import);
2799         up_read(&obd->u.cli.cl_sem);
2800         if (!imp)
2801                 RETURN(-ENODEV);
2802
2803         /* We could possibly pass max_age in the request (as an absolute
2804          * timestamp or a "seconds.usec ago") so the target can avoid doing
2805          * extra calls into the filesystem if that isn't necessary (e.g.
2806          * during mount that would help a bit).  Having relative timestamps
2807          * is not so great if request processing is slow, while absolute
2808          * timestamps are not ideal because they need time synchronization. */
2809         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
2810
2811         class_import_put(imp);
2812
2813         if (req == NULL)
2814                 RETURN(-ENOMEM);
2815
2816         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2817         if (rc) {
2818                 ptlrpc_request_free(req);
2819                 RETURN(rc);
2820         }
2821         ptlrpc_request_set_replen(req);
2822         req->rq_request_portal = OST_CREATE_PORTAL;
2823         ptlrpc_at_set_req_timeout(req);
2824
2825         if (flags & OBD_STATFS_NODELAY) {
2826                 /* procfs requests not want stat in wait for avoid deadlock */
2827                 req->rq_no_resend = 1;
2828                 req->rq_no_delay = 1;
2829         }
2830
2831         rc = ptlrpc_queue_wait(req);
2832         if (rc)
2833                 GOTO(out, rc);
2834
2835         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2836         if (msfs == NULL)
2837                 GOTO(out, rc = -EPROTO);
2838
2839         *osfs = *msfs;
2840
2841         EXIT;
2842 out:
2843         ptlrpc_req_finished(req);
2844         return rc;
2845 }
2846
2847 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2848                          void *karg, void __user *uarg)
2849 {
2850         struct obd_device *obd = exp->exp_obd;
2851         struct obd_ioctl_data *data = karg;
2852         int rc = 0;
2853
2854         ENTRY;
2855         if (!try_module_get(THIS_MODULE)) {
2856                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
2857                        module_name(THIS_MODULE));
2858                 return -EINVAL;
2859         }
2860         switch (cmd) {
2861         case OBD_IOC_CLIENT_RECOVER:
2862                 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
2863                                            data->ioc_inlbuf1, 0);
2864                 if (rc > 0)
2865                         rc = 0;
2866                 break;
2867         case IOC_OSC_SET_ACTIVE:
2868                 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
2869                                               data->ioc_offset);
2870                 break;
2871         default:
2872                 rc = -ENOTTY;
2873                 CDEBUG(D_INODE, "%s: unrecognised ioctl %#x by %s: rc = %d\n",
2874                        obd->obd_name, cmd, current_comm(), rc);
2875                 break;
2876         }
2877
2878         module_put(THIS_MODULE);
2879         return rc;
2880 }
2881
2882 int osc_set_info_async(const struct lu_env *env, struct obd_export *exp,
2883                        u32 keylen, void *key, u32 vallen, void *val,
2884                        struct ptlrpc_request_set *set)
2885 {
2886         struct ptlrpc_request *req;
2887         struct obd_device     *obd = exp->exp_obd;
2888         struct obd_import     *imp = class_exp2cliimp(exp);
2889         char                  *tmp;
2890         int                    rc;
2891         ENTRY;
2892
2893         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
2894
2895         if (KEY_IS(KEY_CHECKSUM)) {
2896                 if (vallen != sizeof(int))
2897                         RETURN(-EINVAL);
2898                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
2899                 RETURN(0);
2900         }
2901
2902         if (KEY_IS(KEY_SPTLRPC_CONF)) {
2903                 sptlrpc_conf_client_adapt(obd);
2904                 RETURN(0);
2905         }
2906
2907         if (KEY_IS(KEY_FLUSH_CTX)) {
2908                 sptlrpc_import_flush_my_ctx(imp);
2909                 RETURN(0);
2910         }
2911
2912         if (KEY_IS(KEY_CACHE_LRU_SHRINK)) {
2913                 struct client_obd *cli = &obd->u.cli;
2914                 long nr = atomic_long_read(&cli->cl_lru_in_list) >> 1;
2915                 long target = *(long *)val;
2916
2917                 nr = osc_lru_shrink(env, cli, min(nr, target), true);
2918                 *(long *)val -= nr;
2919                 RETURN(0);
2920         }
2921
2922         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
2923                 RETURN(-EINVAL);
2924
2925         /* We pass all other commands directly to OST. Since nobody calls osc
2926            methods directly and everybody is supposed to go through LOV, we
2927            assume lov checked invalid values for us.
2928            The only recognised values so far are evict_by_nid and mds_conn.
2929            Even if something bad goes through, we'd get a -EINVAL from OST
2930            anyway. */
2931
2932         req = ptlrpc_request_alloc(imp, KEY_IS(KEY_GRANT_SHRINK) ?
2933                                                 &RQF_OST_SET_GRANT_INFO :
2934                                                 &RQF_OBD_SET_INFO);
2935         if (req == NULL)
2936                 RETURN(-ENOMEM);
2937
2938         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
2939                              RCL_CLIENT, keylen);
2940         if (!KEY_IS(KEY_GRANT_SHRINK))
2941                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
2942                                      RCL_CLIENT, vallen);
2943         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
2944         if (rc) {
2945                 ptlrpc_request_free(req);
2946                 RETURN(rc);
2947         }
2948
2949         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2950         memcpy(tmp, key, keylen);
2951         tmp = req_capsule_client_get(&req->rq_pill, KEY_IS(KEY_GRANT_SHRINK) ?
2952                                                         &RMF_OST_BODY :
2953                                                         &RMF_SETINFO_VAL);
2954         memcpy(tmp, val, vallen);
2955
2956         if (KEY_IS(KEY_GRANT_SHRINK)) {
2957                 struct osc_grant_args *aa;
2958                 struct obdo *oa;
2959
2960                 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2961                 aa = ptlrpc_req_async_args(req);
2962                 OBD_SLAB_ALLOC_PTR_GFP(oa, osc_obdo_kmem, GFP_NOFS);
2963                 if (!oa) {
2964                         ptlrpc_req_finished(req);
2965                         RETURN(-ENOMEM);
2966                 }
2967                 *oa = ((struct ost_body *)val)->oa;
2968                 aa->aa_oa = oa;
2969                 req->rq_interpret_reply = osc_shrink_grant_interpret;
2970         }
2971
2972         ptlrpc_request_set_replen(req);
2973         if (!KEY_IS(KEY_GRANT_SHRINK)) {
2974                 LASSERT(set != NULL);
2975                 ptlrpc_set_add_req(set, req);
2976                 ptlrpc_check_set(NULL, set);
2977         } else {
2978                 ptlrpcd_add_req(req);
2979         }
2980
2981         RETURN(0);
2982 }
2983 EXPORT_SYMBOL(osc_set_info_async);
2984
2985 int osc_reconnect(const struct lu_env *env, struct obd_export *exp,
2986                   struct obd_device *obd, struct obd_uuid *cluuid,
2987                   struct obd_connect_data *data, void *localdata)
2988 {
2989         struct client_obd *cli = &obd->u.cli;
2990
2991         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
2992                 long lost_grant;
2993                 long grant;
2994
2995                 spin_lock(&cli->cl_loi_list_lock);
2996                 grant = cli->cl_avail_grant + cli->cl_reserved_grant;
2997                 if (data->ocd_connect_flags & OBD_CONNECT_GRANT_PARAM) {
2998                         /* restore ocd_grant_blkbits as client page bits */
2999                         data->ocd_grant_blkbits = PAGE_SHIFT;
3000                         grant += cli->cl_dirty_grant;
3001                 } else {
3002                         grant += cli->cl_dirty_pages << PAGE_SHIFT;
3003                 }
3004                 data->ocd_grant = grant ? : 2 * cli_brw_size(obd);
3005                 lost_grant = cli->cl_lost_grant;
3006                 cli->cl_lost_grant = 0;
3007                 spin_unlock(&cli->cl_loi_list_lock);
3008
3009                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: %#llx ocd_version: %d"
3010                        " ocd_grant: %d, lost: %ld.\n", data->ocd_connect_flags,
3011                        data->ocd_version, data->ocd_grant, lost_grant);
3012         }
3013
3014         RETURN(0);
3015 }
3016 EXPORT_SYMBOL(osc_reconnect);
3017
3018 int osc_disconnect(struct obd_export *exp)
3019 {
3020         struct obd_device *obd = class_exp2obd(exp);
3021         int rc;
3022
3023         rc = client_disconnect_export(exp);
3024         /**
3025          * Initially we put del_shrink_grant before disconnect_export, but it
3026          * causes the following problem if setup (connect) and cleanup
3027          * (disconnect) are tangled together.
3028          *      connect p1                     disconnect p2
3029          *   ptlrpc_connect_import
3030          *     ...............               class_manual_cleanup
3031          *                                     osc_disconnect
3032          *                                     del_shrink_grant
3033          *   ptlrpc_connect_interrupt
3034          *     osc_init_grant
3035          *   add this client to shrink list
3036          *                                      cleanup_osc
3037          * Bang! grant shrink thread trigger the shrink. BUG18662
3038          */
3039         osc_del_grant_list(&obd->u.cli);
3040         return rc;
3041 }
3042 EXPORT_SYMBOL(osc_disconnect);
3043
3044 int osc_ldlm_resource_invalidate(struct cfs_hash *hs, struct cfs_hash_bd *bd,
3045                                  struct hlist_node *hnode, void *arg)
3046 {
3047         struct lu_env *env = arg;
3048         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
3049         struct ldlm_lock *lock;
3050         struct osc_object *osc = NULL;
3051         ENTRY;
3052
3053         lock_res(res);
3054         list_for_each_entry(lock, &res->lr_granted, l_res_link) {
3055                 if (lock->l_ast_data != NULL && osc == NULL) {
3056                         osc = lock->l_ast_data;
3057                         cl_object_get(osc2cl(osc));
3058                 }
3059
3060                 /* clear LDLM_FL_CLEANED flag to make sure it will be canceled
3061                  * by the 2nd round of ldlm_namespace_clean() call in
3062                  * osc_import_event(). */
3063                 ldlm_clear_cleaned(lock);
3064         }
3065         unlock_res(res);
3066
3067         if (osc != NULL) {
3068                 osc_object_invalidate(env, osc);
3069                 cl_object_put(env, osc2cl(osc));
3070         }
3071
3072         RETURN(0);
3073 }
3074 EXPORT_SYMBOL(osc_ldlm_resource_invalidate);
3075
3076 static int osc_import_event(struct obd_device *obd,
3077                             struct obd_import *imp,
3078                             enum obd_import_event event)
3079 {
3080         struct client_obd *cli;
3081         int rc = 0;
3082
3083         ENTRY;
3084         LASSERT(imp->imp_obd == obd);
3085
3086         switch (event) {
3087         case IMP_EVENT_DISCON: {
3088                 cli = &obd->u.cli;
3089                 spin_lock(&cli->cl_loi_list_lock);
3090                 cli->cl_avail_grant = 0;
3091                 cli->cl_lost_grant = 0;
3092                 spin_unlock(&cli->cl_loi_list_lock);
3093                 break;
3094         }
3095         case IMP_EVENT_INACTIVE: {
3096                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE);
3097                 break;
3098         }
3099         case IMP_EVENT_INVALIDATE: {
3100                 struct ldlm_namespace *ns = obd->obd_namespace;
3101                 struct lu_env         *env;
3102                 __u16                  refcheck;
3103
3104                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3105
3106                 env = cl_env_get(&refcheck);
3107                 if (!IS_ERR(env)) {
3108                         osc_io_unplug(env, &obd->u.cli, NULL);
3109
3110                         cfs_hash_for_each_nolock(ns->ns_rs_hash,
3111                                                  osc_ldlm_resource_invalidate,
3112                                                  env, 0);
3113                         cl_env_put(env, &refcheck);
3114
3115                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3116                 } else
3117                         rc = PTR_ERR(env);
3118                 break;
3119         }
3120         case IMP_EVENT_ACTIVE: {
3121                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE);
3122                 break;
3123         }
3124         case IMP_EVENT_OCD: {
3125                 struct obd_connect_data *ocd = &imp->imp_connect_data;
3126
3127                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
3128                         osc_init_grant(&obd->u.cli, ocd);
3129
3130                 /* See bug 7198 */
3131                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
3132                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
3133
3134                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD);
3135                 break;
3136         }
3137         case IMP_EVENT_DEACTIVATE: {
3138                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DEACTIVATE);
3139                 break;
3140         }
3141         case IMP_EVENT_ACTIVATE: {
3142                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVATE);
3143                 break;
3144         }
3145         default:
3146                 CERROR("Unknown import event %d\n", event);
3147                 LBUG();
3148         }
3149         RETURN(rc);
3150 }
3151
3152 /**
3153  * Determine whether the lock can be canceled before replaying the lock
3154  * during recovery, see bug16774 for detailed information.