Whamcloud - gitweb
LU-10391 ptlrpc: change rq_self to struct lnet_nid
[fs/lustre-release.git] / lustre / ptlrpc / niobuf.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) 2012, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  */
31
32 #define DEBUG_SUBSYSTEM S_RPC
33 #include <libcfs/linux/linux-mem.h>
34 #include <obd_support.h>
35 #include <lustre_net.h>
36 #include <lustre_lib.h>
37 #include <obd.h>
38 #include <obd_class.h>
39 #include "ptlrpc_internal.h"
40 #include <lnet/lib-lnet.h> /* for CFS_FAIL_PTLRPC_OST_BULK_CB2 */
41
42 /**
43  * Helper function. Sends \a len bytes from \a base at offset \a offset
44  * over \a conn connection to portal \a portal.
45  * Returns 0 on success or error code.
46  */
47 static int ptl_send_buf(struct lnet_handle_md *mdh, void *base, int len,
48                         enum lnet_ack_req ack, struct ptlrpc_cb_id *cbid,
49                         lnet_nid_t self4, struct lnet_process_id peer_id4,
50                         int portal, __u64 xid, unsigned int offset,
51                         struct lnet_handle_md *bulk_cookie)
52 {
53         int rc;
54         struct lnet_md md;
55         struct lnet_nid self;
56         struct lnet_processid peer_id;
57         ENTRY;
58
59         lnet_nid4_to_nid(self4, &self);
60         lnet_pid4_to_pid(peer_id4, &peer_id);
61
62         LASSERT(portal != 0);
63         CDEBUG(D_INFO, "peer_id %s\n", libcfs_id2str(peer_id4));
64         md.start     = base;
65         md.length    = len;
66         md.threshold = (ack == LNET_ACK_REQ) ? 2 : 1;
67         md.options   = PTLRPC_MD_OPTIONS;
68         md.user_ptr  = cbid;
69         md.handler   = ptlrpc_handler;
70         LNetInvalidateMDHandle(&md.bulk_handle);
71
72         if (bulk_cookie) {
73                 md.bulk_handle = *bulk_cookie;
74                 md.options |= LNET_MD_BULK_HANDLE;
75         }
76
77         if (unlikely(ack == LNET_ACK_REQ &&
78                      OBD_FAIL_CHECK_ORSET(OBD_FAIL_PTLRPC_ACK, OBD_FAIL_ONCE))){
79                 /* don't ask for the ack to simulate failing client */
80                 ack = LNET_NOACK_REQ;
81         }
82
83         rc = LNetMDBind(&md, LNET_UNLINK, mdh);
84         if (unlikely(rc != 0)) {
85                 CERROR ("LNetMDBind failed: %d\n", rc);
86                 LASSERT (rc == -ENOMEM);
87                 RETURN (-ENOMEM);
88         }
89
90         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %lld, offset %u\n",
91                len, portal, xid, offset);
92
93         percpu_ref_get(&ptlrpc_pending);
94
95         rc = LNetPut(&self, *mdh, ack,
96                      &peer_id, portal, xid, offset, 0);
97         if (unlikely(rc != 0)) {
98                 int rc2;
99                 /* We're going to get an UNLINK event when I unlink below,
100                  * which will complete just like any other failed send, so
101                  * I fall through and return success here! */
102                 CERROR("LNetPut(%s, %d, %lld) failed: %d\n",
103                        libcfs_id2str(peer_id4), portal, xid, rc);
104                 rc2 = LNetMDUnlink(*mdh);
105                 LASSERTF(rc2 == 0, "rc2 = %d\n", rc2);
106         }
107
108         RETURN (0);
109 }
110
111 #define mdunlink_iterate_helper(mds, count) \
112                 __mdunlink_iterate_helper(mds, count, false) 
113 static void __mdunlink_iterate_helper(struct lnet_handle_md *bd_mds,
114                                       int count, bool discard)
115 {
116         int i;
117
118         for (i = 0; i < count; i++)
119                 __LNetMDUnlink(bd_mds[i], discard);
120 }
121
122 #ifdef HAVE_SERVER_SUPPORT
123 /**
124  * Prepare bulk descriptor for specified incoming request \a req that
125  * can fit \a nfrags * pages. \a type is bulk type. \a portal is where
126  * the bulk to be sent. Used on server-side after request was already
127  * received.
128  * Returns pointer to newly allocatrd initialized bulk descriptor or NULL on
129  * error.
130  */
131 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp(struct ptlrpc_request *req,
132                                               unsigned nfrags, unsigned max_brw,
133                                               unsigned int type,
134                                               unsigned portal,
135                                               const struct ptlrpc_bulk_frag_ops
136                                                 *ops)
137 {
138         struct obd_export *exp = req->rq_export;
139         struct ptlrpc_bulk_desc *desc;
140
141         ENTRY;
142         LASSERT(ptlrpc_is_bulk_op_active(type));
143
144         desc = ptlrpc_new_bulk(nfrags, max_brw, type, portal, ops);
145         if (desc == NULL)
146                 RETURN(NULL);
147
148         desc->bd_export = class_export_get(exp);
149         desc->bd_req = req;
150
151         desc->bd_cbid.cbid_fn  = server_bulk_callback;
152         desc->bd_cbid.cbid_arg = desc;
153
154         /* NB we don't assign rq_bulk here; server-side requests are
155          * re-used, and the handler frees the bulk desc explicitly. */
156
157         return desc;
158 }
159 EXPORT_SYMBOL(ptlrpc_prep_bulk_exp);
160
161 /**
162  * Starts bulk transfer for descriptor \a desc on the server.
163  * Returns 0 on success or error code.
164  */
165 int ptlrpc_start_bulk_transfer(struct ptlrpc_bulk_desc *desc)
166 {
167         struct obd_export       *exp = desc->bd_export;
168         struct lnet_nid          self_nid;
169         struct lnet_processid    peer_id;
170         int                      rc = 0;
171         __u64                    mbits;
172         int                      posted_md;
173         int                      total_md;
174         struct lnet_md           md;
175         ENTRY;
176
177         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_BULK_PUT_NET))
178                 RETURN(0);
179
180         /* NB no locking required until desc is on the network */
181         LASSERT(ptlrpc_is_bulk_op_active(desc->bd_type));
182
183         LASSERT(desc->bd_cbid.cbid_fn == server_bulk_callback);
184         LASSERT(desc->bd_cbid.cbid_arg == desc);
185
186         /*
187          * Multi-Rail: get the preferred self and peer NIDs from the
188          * request, so they are based on the route taken by the
189          * message.
190          */
191         self_nid = desc->bd_req->rq_self;
192         lnet_pid4_to_pid(desc->bd_req->rq_source, &peer_id);
193
194         /* NB total length may be 0 for a read past EOF, so we send 0
195          * length bulks, since the client expects bulk events.
196          *
197          * The client may not need all of the bulk mbits for the RPC. The RPC
198          * used the mbits of the highest bulk mbits needed, and the server masks
199          * off high bits to get bulk count for this RPC. LU-1431 */
200         mbits = desc->bd_req->rq_mbits & ~((__u64)desc->bd_md_max_brw - 1);
201         total_md = desc->bd_req->rq_mbits - mbits + 1;
202         desc->bd_refs = total_md;
203         desc->bd_failure = 0;
204
205         md.user_ptr = &desc->bd_cbid;
206         md.handler = ptlrpc_handler;
207         md.threshold = 2; /* SENT and ACK/REPLY */
208
209         for (posted_md = 0; posted_md < total_md; mbits++) {
210                 md.options = PTLRPC_MD_OPTIONS;
211
212                 /* NB it's assumed that source and sink buffer frags are
213                  * page-aligned. Otherwise we'd have to send client bulk
214                  * sizes over and split server buffer accordingly */
215                 ptlrpc_fill_bulk_md(&md, desc, posted_md);
216                 rc = LNetMDBind(&md, LNET_UNLINK, &desc->bd_mds[posted_md]);
217                 if (rc != 0) {
218                         CERROR("%s: LNetMDBind failed for MD %u: rc = %d\n",
219                                exp->exp_obd->obd_name, posted_md, rc);
220                         LASSERT(rc == -ENOMEM);
221                         if (posted_md == 0) {
222                                 desc->bd_md_count = 0;
223                                 RETURN(-ENOMEM);
224                         }
225                         break;
226                 }
227                 percpu_ref_get(&ptlrpc_pending);
228
229                 /* sanity.sh 224c: lets skip last md */
230                 if (posted_md == desc->bd_md_max_brw - 1)
231                         OBD_FAIL_CHECK_RESET(OBD_FAIL_PTLRPC_CLIENT_BULK_CB3,
232                                              CFS_FAIL_PTLRPC_OST_BULK_CB2);
233
234                 /* Network is about to get at the memory */
235                 if (ptlrpc_is_bulk_put_source(desc->bd_type))
236                         rc = LNetPut(&self_nid, desc->bd_mds[posted_md],
237                                      LNET_ACK_REQ, &peer_id,
238                                      desc->bd_portal, mbits, 0, 0);
239                 else
240                         rc = LNetGet(&self_nid, desc->bd_mds[posted_md],
241                                      &peer_id, desc->bd_portal,
242                                      mbits, 0, false);
243
244                 posted_md++;
245                 if (rc != 0) {
246                         CERROR("%s: failed bulk transfer with %s:%u x%llu: "
247                                "rc = %d\n", exp->exp_obd->obd_name,
248                                libcfs_idstr(&peer_id), desc->bd_portal,
249                                mbits, rc);
250                         break;
251                 }
252         }
253
254         if (rc != 0) {
255                 /* Can't send, so we unlink the MD bound above.  The UNLINK
256                  * event this creates will signal completion with failure,
257                  * so we return SUCCESS here! */
258                 spin_lock(&desc->bd_lock);
259                 desc->bd_refs -= total_md - posted_md;
260                 spin_unlock(&desc->bd_lock);
261                 LASSERT(desc->bd_refs >= 0);
262
263                 mdunlink_iterate_helper(desc->bd_mds, posted_md);
264                 RETURN(0);
265         }
266
267         CDEBUG(D_NET, "Transferring %u pages %u bytes via portal %d "
268                "id %s mbits %#llx-%#llx\n", desc->bd_iov_count,
269                desc->bd_nob, desc->bd_portal, libcfs_idstr(&peer_id),
270                mbits - posted_md, mbits - 1);
271
272         RETURN(0);
273 }
274
275 /**
276  * Server side bulk abort. Idempotent. Not thread-safe (i.e. only
277  * serialises with completion callback)
278  */
279 void ptlrpc_abort_bulk(struct ptlrpc_bulk_desc *desc)
280 {
281         LASSERT(!in_interrupt());           /* might sleep */
282
283         if (!ptlrpc_server_bulk_active(desc))   /* completed or */
284                 return;                         /* never started */
285
286         /* We used to poison the pages with 0xab here because we did not want to
287          * send any meaningful data over the wire for evicted clients (bug 9297)
288          * However, this is no longer safe now that we use the page cache on the
289          * OSS (bug 20560) */
290
291         /* The unlink ensures the callback happens ASAP and is the last
292          * one.  If it fails, it must be because completion just happened,
293          * but we must still wait_event_idle_timeout() in this case, to give
294          * us a chance to run server_bulk_callback()
295          */
296         __mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw, true);
297
298         for (;;) {
299                 /* Network access will complete in finite time but the HUGE
300                  * timeout lets us CWARN for visibility of sluggish NALs */
301                 int seconds = PTLRPC_REQ_LONG_UNLINK;
302
303                 while (seconds > 0 &&
304                        wait_event_idle_timeout(desc->bd_waitq,
305                                                !ptlrpc_server_bulk_active(desc),
306                                                cfs_time_seconds(1)) == 0)
307                         seconds -= 1;
308                 if (seconds > 0)
309                         return;
310
311                 CWARN("Unexpectedly long timeout: desc %p\n", desc);
312         }
313 }
314 #endif /* HAVE_SERVER_SUPPORT */
315
316 /**
317  * Register bulk at the sender for later transfer.
318  * Returns 0 on success or error code.
319  */
320 int ptlrpc_register_bulk(struct ptlrpc_request *req)
321 {
322         struct ptlrpc_bulk_desc *desc = req->rq_bulk;
323         struct lnet_processid peer;
324         int rc = 0;
325         int posted_md;
326         int total_md;
327         __u64 mbits;
328         struct lnet_me *me;
329         struct lnet_md md;
330         ENTRY;
331
332         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_BULK_GET_NET))
333                 RETURN(0);
334
335         /* NB no locking required until desc is on the network */
336         LASSERT(desc->bd_nob > 0);
337         LASSERT(desc->bd_md_max_brw <= PTLRPC_BULK_OPS_COUNT);
338         LASSERT(desc->bd_iov_count <= PTLRPC_MAX_BRW_PAGES);
339         LASSERT(desc->bd_req != NULL);
340         LASSERT(ptlrpc_is_bulk_op_passive(desc->bd_type));
341
342         /* cleanup the state of the bulk for it will be reused */
343         if (req->rq_resend || req->rq_send_state == LUSTRE_IMP_REPLAY)
344                 desc->bd_nob_transferred = 0;
345         else if (desc->bd_nob_transferred != 0)
346                 /* If the network failed after an RPC was sent, this condition
347                  * could happen.  Rather than assert (was here before), return
348                  * an EIO error. */
349                 RETURN(-EIO);
350
351         desc->bd_failure = 0;
352
353         peer = desc->bd_import->imp_connection->c_peer;
354
355         LASSERT(desc->bd_cbid.cbid_fn == client_bulk_callback);
356         LASSERT(desc->bd_cbid.cbid_arg == desc);
357
358         total_md = desc->bd_md_count;
359         /* rq_mbits is matchbits of the final bulk */
360         mbits = req->rq_mbits - desc->bd_md_count + 1;
361
362         LASSERTF(mbits == (req->rq_mbits & PTLRPC_BULK_OPS_MASK),
363                  "first mbits = x%llu, last mbits = x%llu\n",
364                  mbits, req->rq_mbits);
365         LASSERTF(!(desc->bd_registered &&
366                    req->rq_send_state != LUSTRE_IMP_REPLAY) ||
367                  mbits != desc->bd_last_mbits,
368                  "registered: %d  rq_mbits: %llu bd_last_mbits: %llu\n",
369                  desc->bd_registered, mbits, desc->bd_last_mbits);
370
371         desc->bd_registered = 1;
372         desc->bd_last_mbits = mbits;
373         desc->bd_refs = total_md;
374         md.user_ptr = &desc->bd_cbid;
375         md.handler = ptlrpc_handler;
376         md.threshold = 1;                       /* PUT or GET */
377
378         for (posted_md = 0; posted_md < desc->bd_md_count;
379              posted_md++, mbits++) {
380                 md.options = PTLRPC_MD_OPTIONS |
381                              (ptlrpc_is_bulk_op_get(desc->bd_type) ?
382                               LNET_MD_OP_GET : LNET_MD_OP_PUT);
383                 ptlrpc_fill_bulk_md(&md, desc, posted_md);
384
385                 if (posted_md > 0 && posted_md + 1 == desc->bd_md_count &&
386                     OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_BULK_ATTACH)) {
387                         rc = -ENOMEM;
388                 } else {
389                         me = LNetMEAttach(desc->bd_portal, &peer, mbits, 0,
390                                   LNET_UNLINK, LNET_INS_AFTER);
391                         rc = PTR_ERR_OR_ZERO(me);
392                 }
393                 if (rc != 0) {
394                         CERROR("%s: LNetMEAttach failed x%llu/%d: rc = %d\n",
395                                desc->bd_import->imp_obd->obd_name, mbits,
396                                posted_md, rc);
397                         break;
398                 }
399                 percpu_ref_get(&ptlrpc_pending);
400
401                 /* About to let the network at it... */
402                 rc = LNetMDAttach(me, &md, LNET_UNLINK,
403                                   &desc->bd_mds[posted_md]);
404                 if (rc != 0) {
405                         CERROR("%s: LNetMDAttach failed x%llu/%d: rc = %d\n",
406                                desc->bd_import->imp_obd->obd_name, mbits,
407                                posted_md, rc);
408                         break;
409                 }
410         }
411
412         if (rc != 0) {
413                 LASSERT(rc == -ENOMEM);
414                 spin_lock(&desc->bd_lock);
415                 desc->bd_refs -= total_md - posted_md;
416                 spin_unlock(&desc->bd_lock);
417                 LASSERT(desc->bd_refs >= 0);
418                 mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw);
419                 req->rq_status = -ENOMEM;
420                 desc->bd_registered = 0;
421                 RETURN(-ENOMEM);
422         }
423
424         spin_lock(&desc->bd_lock);
425         /* Holler if peer manages to touch buffers before he knows the mbits */
426         if (desc->bd_refs != total_md)
427                 CWARN("%s: Peer %s touched %d buffers while I registered\n",
428                       desc->bd_import->imp_obd->obd_name, libcfs_idstr(&peer),
429                       total_md - desc->bd_refs);
430         spin_unlock(&desc->bd_lock);
431
432         CDEBUG(D_NET,
433                "Setup %u bulk %s buffers: %u pages %u bytes, mbits x%#llx-%#llx, portal %u\n",
434                desc->bd_refs,
435                ptlrpc_is_bulk_op_get(desc->bd_type) ? "get-source" : "put-sink",
436                desc->bd_iov_count, desc->bd_nob,
437                desc->bd_last_mbits, req->rq_mbits, desc->bd_portal);
438
439         RETURN(0);
440 }
441
442 /**
443  * Disconnect a bulk desc from the network. Idempotent. Not
444  * thread-safe (i.e. only interlocks with completion callback).
445  * Returns 1 on success or 0 if network unregistration failed for whatever
446  * reason.
447  */
448 int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async)
449 {
450         struct ptlrpc_bulk_desc *desc = req->rq_bulk;
451         ENTRY;
452
453         LASSERT(!in_interrupt());     /* might sleep */
454
455         if (desc)
456                 desc->bd_registered = 0;
457
458         /* Let's setup deadline for reply unlink. */
459         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK) &&
460             async && req->rq_bulk_deadline == 0 && cfs_fail_val == 0)
461                 req->rq_bulk_deadline = ktime_get_real_seconds() +
462                                         PTLRPC_REQ_LONG_UNLINK;
463
464         if (ptlrpc_client_bulk_active(req) == 0)        /* completed or */
465                 RETURN(1);                              /* never registered */
466
467         LASSERT(desc->bd_req == req);  /* bd_req NULL until registered */
468
469         /* the unlink ensures the callback happens ASAP and is the last
470          * one.  If it fails, it must be because completion just happened,
471          * but we must still wait_event_idle_timeout() in this case to give
472          * us a chance to run client_bulk_callback()
473          */
474         mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw);
475
476         if (ptlrpc_client_bulk_active(req) == 0)        /* completed or */
477                 RETURN(1);                              /* never registered */
478
479         /* Move to "Unregistering" phase as bulk was not unlinked yet. */
480         ptlrpc_rqphase_move(req, RQ_PHASE_UNREG_BULK);
481
482         /* Do not wait for unlink to finish. */
483         if (async)
484                 RETURN(0);
485
486         for (;;) {
487                 /* The wq argument is ignored by user-space wait_event macros */
488                 wait_queue_head_t *wq = (req->rq_set != NULL) ?
489                                         &req->rq_set->set_waitq :
490                                         &req->rq_reply_waitq;
491                 /*
492                  * Network access will complete in finite time but the HUGE
493                  * timeout lets us CWARN for visibility of sluggish NALs.
494                  */
495                 int seconds = PTLRPC_REQ_LONG_UNLINK;
496
497                 while (seconds > 0 &&
498                        wait_event_idle_timeout(*wq,
499                                                !ptlrpc_client_bulk_active(req),
500                                                cfs_time_seconds(1)) == 0)
501                         seconds -= 1;
502                 if (seconds > 0) {
503                         ptlrpc_rqphase_move(req, req->rq_next_phase);
504                         RETURN(1);
505                 }
506
507                 DEBUG_REQ(D_WARNING, req, "Unexpectedly long timeout: desc %p",
508                           desc);
509         }
510         RETURN(0);
511 }
512
513 static void ptlrpc_at_set_reply(struct ptlrpc_request *req, int flags)
514 {
515         struct ptlrpc_service_part      *svcpt = req->rq_rqbd->rqbd_svcpt;
516         struct ptlrpc_service           *svc = svcpt->scp_service;
517         timeout_t service_timeout;
518
519         service_timeout = clamp_t(timeout_t, ktime_get_real_seconds() -
520                                              req->rq_arrival_time.tv_sec, 1,
521                                   (AT_OFF ? obd_timeout * 3 / 2 : at_max));
522         if (!(flags & PTLRPC_REPLY_EARLY) &&
523             (req->rq_type != PTL_RPC_MSG_ERR) &&
524             (req->rq_reqmsg != NULL) &&
525             !(lustre_msg_get_flags(req->rq_reqmsg) &
526               (MSG_RESENT | MSG_REPLAY |
527                MSG_REQ_REPLAY_DONE | MSG_LOCK_REPLAY_DONE))) {
528                 /* early replies, errors and recovery requests don't count
529                  * toward our service time estimate
530                  */
531                 timeout_t oldse = at_measured(&svcpt->scp_at_estimate,
532                                               service_timeout);
533
534                 if (oldse != 0) {
535                         DEBUG_REQ(D_ADAPTTO, req,
536                                   "svc %s changed estimate from %d to %d",
537                                   svc->srv_name, oldse,
538                                   at_get(&svcpt->scp_at_estimate));
539                 }
540         }
541         /* Report actual service time for client latency calc */
542         lustre_msg_set_service_timeout(req->rq_repmsg, service_timeout);
543         /* Report service time estimate for future client reqs, but report 0
544          * (to be ignored by client) if it's an error reply during recovery.
545          * b=15815
546          */
547         if (req->rq_type == PTL_RPC_MSG_ERR &&
548             (req->rq_export == NULL ||
549              req->rq_export->exp_obd->obd_recovering)) {
550                 lustre_msg_set_timeout(req->rq_repmsg, 0);
551         } else {
552                 timeout_t timeout;
553
554                 if (req->rq_export && req->rq_reqmsg != NULL &&
555                     (flags & PTLRPC_REPLY_EARLY) &&
556                     lustre_msg_get_flags(req->rq_reqmsg) &
557                     (MSG_REPLAY | MSG_REQ_REPLAY_DONE | MSG_LOCK_REPLAY_DONE)) {
558                         struct obd_device *exp_obd = req->rq_export->exp_obd;
559
560                         timeout = ktime_get_real_seconds() -
561                                   req->rq_arrival_time.tv_sec +
562                                   min_t(timeout_t, at_extra,
563                                         exp_obd->obd_recovery_timeout / 4);
564                 } else {
565                         timeout = at_get(&svcpt->scp_at_estimate);
566                 }
567                 lustre_msg_set_timeout(req->rq_repmsg, timeout);
568         }
569
570         if (req->rq_reqmsg &&
571             !(lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT)) {
572                 CDEBUG(D_ADAPTTO, "No early reply support: flags=%#x "
573                        "req_flags=%#x magic=%x/%x len=%d\n",
574                        flags, lustre_msg_get_flags(req->rq_reqmsg),
575                        lustre_msg_get_magic(req->rq_reqmsg),
576                        lustre_msg_get_magic(req->rq_repmsg), req->rq_replen);
577         }
578 }
579
580 /**
581  * Send request reply from request \a req reply buffer.
582  * \a flags defines reply types
583  * Returns 0 on success or error code
584  */
585 int ptlrpc_send_reply(struct ptlrpc_request *req, int flags)
586 {
587         struct ptlrpc_reply_state *rs = req->rq_reply_state;
588         struct ptlrpc_connection  *conn;
589         int                        rc;
590
591         /* We must already have a reply buffer (only ptlrpc_error() may be
592          * called without one). The reply generated by sptlrpc layer (e.g.
593          * error notify, etc.) might have NULL rq->reqmsg; Otherwise we must
594          * have a request buffer which is either the actual (swabbed) incoming
595          * request, or a saved copy if this is a req saved in
596          * target_queue_final_reply().
597          */
598         LASSERT (req->rq_no_reply == 0);
599         LASSERT (req->rq_reqbuf != NULL);
600         LASSERT (rs != NULL);
601         LASSERT ((flags & PTLRPC_REPLY_MAYBE_DIFFICULT) || !rs->rs_difficult);
602         LASSERT (req->rq_repmsg != NULL);
603         LASSERT (req->rq_repmsg == rs->rs_msg);
604         LASSERT (rs->rs_cb_id.cbid_fn == reply_out_callback);
605         LASSERT (rs->rs_cb_id.cbid_arg == rs);
606
607         /* There may be no rq_export during failover */
608
609         if (unlikely(req->rq_export && req->rq_export->exp_obd &&
610                      req->rq_export->exp_obd->obd_fail)) {
611                 /* Failed obd's only send ENODEV */
612                 req->rq_type = PTL_RPC_MSG_ERR;
613                 req->rq_status = -ENODEV;
614                 CDEBUG(D_HA, "sending ENODEV from failed obd %d\n",
615                        req->rq_export->exp_obd->obd_minor);
616         }
617
618         if (req->rq_type != PTL_RPC_MSG_ERR)
619                 req->rq_type = PTL_RPC_MSG_REPLY;
620
621         lustre_msg_set_type(req->rq_repmsg, req->rq_type);
622         lustre_msg_set_status(req->rq_repmsg,
623                               ptlrpc_status_hton(req->rq_status));
624         lustre_msg_set_opc(req->rq_repmsg,
625                 req->rq_reqmsg ? lustre_msg_get_opc(req->rq_reqmsg) : 0);
626
627         target_pack_pool_reply(req);
628
629         ptlrpc_at_set_reply(req, flags);
630
631         if (req->rq_export == NULL || req->rq_export->exp_connection == NULL)
632                 conn = ptlrpc_connection_get(req->rq_peer, &req->rq_self, NULL);
633         else
634                 conn = ptlrpc_connection_addref(req->rq_export->exp_connection);
635
636         if (unlikely(conn == NULL)) {
637                 CERROR("not replying on NULL connection\n"); /* bug 9635 */
638                 return -ENOTCONN;
639         }
640         ptlrpc_rs_addref(rs);                   /* +1 ref for the network */
641
642         rc = sptlrpc_svc_wrap_reply(req);
643         if (unlikely(rc))
644                 goto out;
645
646         req->rq_sent = ktime_get_real_seconds();
647
648         rc = ptl_send_buf(&rs->rs_md_h, rs->rs_repbuf, rs->rs_repdata_len,
649                           (rs->rs_difficult && !rs->rs_no_ack) ?
650                           LNET_ACK_REQ : LNET_NOACK_REQ,
651                           &rs->rs_cb_id, lnet_nid_to_nid4(&req->rq_self),
652                           req->rq_source,
653                           ptlrpc_req2svc(req)->srv_rep_portal,
654                           req->rq_rep_mbits ? req->rq_rep_mbits : req->rq_xid,
655                           req->rq_reply_off, NULL);
656 out:
657         if (unlikely(rc != 0))
658                 ptlrpc_req_drop_rs(req);
659         ptlrpc_connection_put(conn);
660         return rc;
661 }
662
663 int ptlrpc_reply (struct ptlrpc_request *req)
664 {
665         if (req->rq_no_reply)
666                 return 0;
667         else
668                 return (ptlrpc_send_reply(req, 0));
669 }
670
671 /**
672  * For request \a req send an error reply back. Create empty
673  * reply buffers if necessary.
674  */
675 int ptlrpc_send_error(struct ptlrpc_request *req, int may_be_difficult)
676 {
677         int rc;
678         ENTRY;
679
680         if (req->rq_no_reply)
681                 RETURN(0);
682
683         if (!req->rq_repmsg) {
684                 rc = lustre_pack_reply(req, 1, NULL, NULL);
685                 if (rc)
686                         RETURN(rc);
687         }
688
689         if (req->rq_status != -ENOSPC && req->rq_status != -EACCES &&
690             req->rq_status != -EPERM && req->rq_status != -ENOENT &&
691             req->rq_status != -EINPROGRESS && req->rq_status != -EDQUOT &&
692             req->rq_status != -EROFS)
693                 req->rq_type = PTL_RPC_MSG_ERR;
694
695         rc = ptlrpc_send_reply(req, may_be_difficult);
696         RETURN(rc);
697 }
698
699 int ptlrpc_error(struct ptlrpc_request *req)
700 {
701         return ptlrpc_send_error(req, 0);
702 }
703
704 /**
705  * Send request \a request.
706  * if \a noreply is set, don't expect any reply back and don't set up
707  * reply buffers.
708  * Returns 0 on success or error code.
709  */
710 int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
711 {
712         int rc;
713         __u32 opc;
714         int mpflag = 0;
715         bool rep_mbits = false;
716         struct lnet_handle_md bulk_cookie;
717         struct lnet_processid peer;
718         struct ptlrpc_connection *connection;
719         struct lnet_me *reply_me = NULL;
720         struct lnet_md reply_md;
721         struct obd_import *imp = request->rq_import;
722         struct obd_device *obd = imp->imp_obd;
723         ENTRY;
724
725         LNetInvalidateMDHandle(&bulk_cookie);
726
727         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_RPC))
728                 RETURN(0);
729
730         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DELAY_RECOV) &&
731                      lustre_msg_get_opc(request->rq_reqmsg) == MDS_CONNECT &&
732                      strcmp(obd->obd_type->typ_name, LUSTRE_OSP_NAME) == 0)) {
733                 RETURN(0);
734         }
735
736         LASSERT(request->rq_type == PTL_RPC_MSG_REQUEST);
737         LASSERT(request->rq_wait_ctx == 0);
738
739         /* If this is a re-transmit, we're required to have disengaged
740          * cleanly from the previous attempt */
741         LASSERT(!request->rq_receiving_reply);
742         LASSERT(!((lustre_msg_get_flags(request->rq_reqmsg) & MSG_REPLAY) &&
743                   (imp->imp_state == LUSTRE_IMP_FULL)));
744
745         if (unlikely(obd != NULL && obd->obd_fail)) {
746                 CDEBUG(D_HA, "muting rpc for failed imp obd %s\n",
747                        obd->obd_name);
748                 /* this prevents us from waiting in ptlrpc_queue_wait */
749                 spin_lock(&request->rq_lock);
750                 request->rq_err = 1;
751                 spin_unlock(&request->rq_lock);
752                 request->rq_status = -ENODEV;
753                 RETURN(-ENODEV);
754         }
755
756         connection = imp->imp_connection;
757
758         lustre_msg_set_handle(request->rq_reqmsg,
759                               &imp->imp_remote_handle);
760         lustre_msg_set_type(request->rq_reqmsg, PTL_RPC_MSG_REQUEST);
761         lustre_msg_set_conn_cnt(request->rq_reqmsg,
762                                 imp->imp_conn_cnt);
763         lustre_msghdr_set_flags(request->rq_reqmsg,
764                                 imp->imp_msghdr_flags);
765
766         /* If it's the first time to resend the request for EINPROGRESS,
767          * we need to allocate a new XID (see after_reply()), it's different
768          * from the resend for reply timeout. */
769         if (request->rq_nr_resend != 0 &&
770             list_empty(&request->rq_unreplied_list)) {
771                 __u64 min_xid = 0;
772                 /* resend for EINPROGRESS, allocate new xid to avoid reply
773                  * reconstruction */
774                 spin_lock(&imp->imp_lock);
775                 ptlrpc_assign_next_xid_nolock(request);
776                 min_xid = ptlrpc_known_replied_xid(imp);
777                 spin_unlock(&imp->imp_lock);
778
779                 lustre_msg_set_last_xid(request->rq_reqmsg, min_xid);
780                 DEBUG_REQ(D_RPCTRACE, request,
781                           "Allocating new XID for resend on EINPROGRESS");
782         }
783
784         opc = lustre_msg_get_opc(request->rq_reqmsg);
785         if (opc != OST_CONNECT && opc != MDS_CONNECT &&
786             opc != MGS_CONNECT && OCD_HAS_FLAG(&imp->imp_connect_data, FLAGS2))
787                 rep_mbits = imp->imp_connect_data.ocd_connect_flags2 &
788                         OBD_CONNECT2_REP_MBITS;
789
790         if ((request->rq_bulk != NULL) || rep_mbits) {
791                 ptlrpc_set_mbits(request);
792                 lustre_msg_set_mbits(request->rq_reqmsg, request->rq_mbits);
793         }
794
795         if (list_empty(&request->rq_unreplied_list) ||
796             request->rq_xid <= imp->imp_known_replied_xid) {
797                 DEBUG_REQ(D_ERROR, request,
798                           "xid=%llu, replied=%llu, list_empty=%d",
799                           request->rq_xid, imp->imp_known_replied_xid,
800                           list_empty(&request->rq_unreplied_list));
801                 LBUG();
802         }
803
804         /** For enabled AT all request should have AT_SUPPORT in the
805          * FULL import state when OBD_CONNECT_AT is set */
806         LASSERT(AT_OFF || imp->imp_state != LUSTRE_IMP_FULL ||
807                 (imp->imp_msghdr_flags & MSGHDR_AT_SUPPORT) ||
808                 !(imp->imp_connect_data.ocd_connect_flags &
809                   OBD_CONNECT_AT));
810
811         if (request->rq_resend) {
812                 lustre_msg_add_flags(request->rq_reqmsg, MSG_RESENT);
813                 if (request->rq_resend_cb != NULL)
814                         request->rq_resend_cb(request, &request->rq_async_args);
815         }
816         if (request->rq_memalloc)
817                 mpflag = memalloc_noreclaim_save();
818
819         rc = sptlrpc_cli_wrap_request(request);
820         if (rc)
821                 GOTO(out, rc);
822
823         /* bulk register should be done after wrap_request() */
824         if (request->rq_bulk != NULL) {
825                 rc = ptlrpc_register_bulk (request);
826                 if (rc != 0)
827                         GOTO(cleanup_bulk, rc);
828                 /*
829                  * All the mds in the request will have the same cpt
830                  * encoded in the cookie. So we can just get the first
831                  * one.
832                  */
833                 bulk_cookie = request->rq_bulk->bd_mds[0];
834         }
835
836         if (!noreply) {
837                 LASSERT (request->rq_replen != 0);
838                 if (request->rq_repbuf == NULL) {
839                         LASSERT(request->rq_repdata == NULL);
840                         LASSERT(request->rq_repmsg == NULL);
841                         rc = sptlrpc_cli_alloc_repbuf(request,
842                                                       request->rq_replen);
843                         if (rc) {
844                                 /* this prevents us from looping in
845                                  * ptlrpc_queue_wait */
846                                 spin_lock(&request->rq_lock);
847                                 request->rq_err = 1;
848                                 spin_unlock(&request->rq_lock);
849                                 request->rq_status = rc;
850                                 GOTO(cleanup_bulk, rc);
851                         }
852                 } else {
853                         request->rq_repdata = NULL;
854                         request->rq_repmsg = NULL;
855                 }
856
857                 peer = connection->c_peer;
858                 if (request->rq_bulk &&
859                     OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_BULK_REPLY_ATTACH)) {
860                         reply_me = ERR_PTR(-ENOMEM);
861                 } else {
862                         reply_me = LNetMEAttach(request->rq_reply_portal,
863                                                 &peer,
864                                                 rep_mbits ? request->rq_mbits :
865                                                 request->rq_xid,
866                                                 0, LNET_UNLINK, LNET_INS_AFTER);
867                 }
868
869                 if (IS_ERR(reply_me)) {
870                         rc = PTR_ERR(reply_me);
871                         CERROR("LNetMEAttach failed: %d\n", rc);
872                         LASSERT(rc == -ENOMEM);
873                         GOTO(cleanup_bulk, rc = -ENOMEM);
874                 }
875         }
876
877         spin_lock(&request->rq_lock);
878         /* We are responsible for unlinking the reply buffer */
879         request->rq_reply_unlinked = noreply;
880         request->rq_receiving_reply = !noreply;
881         /* Clear any flags that may be present from previous sends. */
882         request->rq_req_unlinked = 0;
883         request->rq_replied = 0;
884         request->rq_err = 0;
885         request->rq_timedout = 0;
886         request->rq_net_err = 0;
887         request->rq_resend = 0;
888         request->rq_restart = 0;
889         request->rq_reply_truncated = 0;
890         spin_unlock(&request->rq_lock);
891
892         if (!noreply) {
893                 reply_md.start     = request->rq_repbuf;
894                 reply_md.length    = request->rq_repbuf_len;
895                 /* Allow multiple early replies */
896                 reply_md.threshold = LNET_MD_THRESH_INF;
897                 /* Manage remote for early replies */
898                 reply_md.options   = PTLRPC_MD_OPTIONS | LNET_MD_OP_PUT |
899                         LNET_MD_MANAGE_REMOTE |
900                         LNET_MD_TRUNCATE; /* allow to make EOVERFLOW error */;
901                 reply_md.user_ptr  = &request->rq_reply_cbid;
902                 reply_md.handler = ptlrpc_handler;
903
904                 /* We must see the unlink callback to set rq_reply_unlinked,
905                  * so we can't auto-unlink */
906                 rc = LNetMDAttach(reply_me, &reply_md, LNET_RETAIN,
907                                   &request->rq_reply_md_h);
908                 if (rc != 0) {
909                         CERROR("LNetMDAttach failed: %d\n", rc);
910                         LASSERT(rc == -ENOMEM);
911                         spin_lock(&request->rq_lock);
912                         /* ...but the MD attach didn't succeed... */
913                         request->rq_receiving_reply = 0;
914                         spin_unlock(&request->rq_lock);
915                         GOTO(cleanup_bulk, rc = -ENOMEM);
916                 }
917                 percpu_ref_get(&ptlrpc_pending);
918
919                 CDEBUG(D_NET,
920                        "Setup reply buffer: %u bytes, xid %llu, portal %u\n",
921                        request->rq_repbuf_len, request->rq_xid,
922                        request->rq_reply_portal);
923         }
924
925         /* add references on request for request_out_callback */
926         ptlrpc_request_addref(request);
927         if (obd != NULL && obd->obd_svc_stats != NULL)
928                 lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQACTIVE_CNTR,
929                                     atomic_read(&imp->imp_inflight));
930
931         OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_DELAY_SEND, request->rq_timeout + 5);
932
933         request->rq_sent_ns = ktime_get_real();
934         request->rq_sent = ktime_get_real_seconds();
935         /* We give the server rq_timeout secs to process the req, and
936          * add the network latency for our local timeout.
937          */
938         request->rq_deadline = request->rq_sent + request->rq_timeout +
939                 ptlrpc_at_get_net_latency(request);
940
941         DEBUG_REQ(D_INFO, request, "send flags=%x",
942                   lustre_msg_get_flags(request->rq_reqmsg));
943         rc = ptl_send_buf(&request->rq_req_md_h,
944                           request->rq_reqbuf, request->rq_reqdata_len,
945                           LNET_NOACK_REQ, &request->rq_req_cbid,
946                           LNET_NID_ANY,
947                           lnet_pid_to_pid4(&connection->c_peer),
948                           request->rq_request_portal,
949                           request->rq_xid, 0, &bulk_cookie);
950         if (likely(rc == 0))
951                 GOTO(out, rc);
952
953         request->rq_req_unlinked = 1;
954         ptlrpc_req_finished(request);
955         if (noreply)
956                 GOTO(out, rc);
957
958         LNetMDUnlink(request->rq_reply_md_h);
959
960         /* UNLINKED callback called synchronously */
961         LASSERT(!request->rq_receiving_reply);
962
963  cleanup_bulk:
964         /* We do sync unlink here as there was no real transfer here so
965          * the chance to have long unlink to sluggish net is smaller here. */
966         ptlrpc_unregister_bulk(request, 0);
967  out:
968         if (rc == -ENOMEM) {
969                 /* set rq_sent so that this request is treated
970                  * as a delayed send in the upper layers */
971                 request->rq_sent = ktime_get_real_seconds();
972         }
973
974         if (request->rq_memalloc)
975                 memalloc_noreclaim_restore(mpflag);
976
977         return rc;
978 }
979 EXPORT_SYMBOL(ptl_send_rpc);
980
981 /**
982  * Register request buffer descriptor for request receiving.
983  */
984 int ptlrpc_register_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
985 {
986         struct ptlrpc_service *service = rqbd->rqbd_svcpt->scp_service;
987         static struct lnet_processid match_id = {
988                 .nid = LNET_ANY_NID,
989                 .pid = LNET_PID_ANY
990         };
991         int rc;
992         struct lnet_md md;
993         struct lnet_me *me;
994
995         CDEBUG(D_NET, "%s: registering portal %d\n", service->srv_name,
996                service->srv_req_portal);
997
998         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_RQBD))
999                 return -ENOMEM;
1000
1001         /* NB: CPT affinity service should use new LNet flag LNET_INS_LOCAL,
1002          * which means buffer can only be attached on local CPT, and LND
1003          * threads can find it by grabbing a local lock */
1004         me = LNetMEAttach(service->srv_req_portal,
1005                           &match_id, 0, ~0, LNET_UNLINK,
1006                           rqbd->rqbd_svcpt->scp_cpt >= 0 ?
1007                           LNET_INS_LOCAL : LNET_INS_AFTER);
1008         if (IS_ERR(me)) {
1009                 CERROR("%s: LNetMEAttach failed: rc = %ld\n",
1010                        service->srv_name, PTR_ERR(me));
1011                 return PTR_ERR(me);
1012         }
1013
1014         LASSERT(rqbd->rqbd_refcount == 0);
1015         rqbd->rqbd_refcount = 1;
1016
1017         md.start     = rqbd->rqbd_buffer;
1018         md.length    = service->srv_buf_size;
1019         md.max_size  = service->srv_max_req_size;
1020         md.threshold = LNET_MD_THRESH_INF;
1021         md.options   = PTLRPC_MD_OPTIONS | LNET_MD_OP_PUT | LNET_MD_MAX_SIZE;
1022         md.user_ptr  = &rqbd->rqbd_cbid;
1023         md.handler   = ptlrpc_handler;
1024
1025         rc = LNetMDAttach(me, &md, LNET_UNLINK, &rqbd->rqbd_md_h);
1026         if (rc == 0) {
1027                 percpu_ref_get(&ptlrpc_pending);
1028                 return 0;
1029         }
1030
1031         CERROR("%s: LNetMDAttach failed: rc = %d\n", service->srv_name, rc);
1032         LASSERT(rc == -ENOMEM);
1033         rqbd->rqbd_refcount = 0;
1034
1035         return rc;
1036 }