Whamcloud - gitweb
eea505abd508422e5201bf7e8ef54d7089ccb50d
[fs/lustre-release.git] / lustre / include / lustre_net.h
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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2010, 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 /** \defgroup PtlRPC Portal RPC and networking module.
33  *
34  * PortalRPC is the layer used by rest of lustre code to achieve network
35  * communications: establish connections with corresponding export and import
36  * states, listen for a service, send and receive RPCs.
37  * PortalRPC also includes base recovery framework: packet resending and
38  * replaying, reconnections, pinger.
39  *
40  * PortalRPC utilizes LNet as its transport layer.
41  *
42  * @{
43  */
44
45
46 #ifndef _LUSTRE_NET_H
47 #define _LUSTRE_NET_H
48
49 /** \defgroup net net
50  *
51  * @{
52  */
53 #include <linux/kobject.h>
54 #include <linux/uio.h>
55 #include <libcfs/libcfs.h>
56 #include <lnet/api.h>
57 #include <lnet/lib-types.h>
58 #include <uapi/linux/lnet/nidstr.h>
59 #include <uapi/linux/lustre/lustre_idl.h>
60 #include <lustre_ha.h>
61 #include <lustre_sec.h>
62 #include <lustre_import.h>
63 #include <lprocfs_status.h>
64 #include <lu_object.h>
65 #include <lustre_req_layout.h>
66 #include <obd_support.h>
67 #include <uapi/linux/lustre/lustre_ver.h>
68
69 /* MD flags we _always_ use */
70 #define PTLRPC_MD_OPTIONS  0
71
72 /**
73  * log2 max # of bulk operations in one request: 2=4MB/RPC, 5=32MB/RPC, ...
74  * In order for the client and server to properly negotiate the maximum
75  * possible transfer size, PTLRPC_BULK_OPS_COUNT must be a power-of-two
76  * value.  The client is free to limit the actual RPC size for any bulk
77  * transfer via cl_max_pages_per_rpc to some non-power-of-two value.
78  * NOTE: This is limited to 16 (=64GB RPCs) by IOOBJ_MAX_BRW_BITS. */
79 #define PTLRPC_BULK_OPS_BITS    6
80 #if PTLRPC_BULK_OPS_BITS > 16
81 #error "More than 65536 BRW RPCs not allowed by IOOBJ_MAX_BRW_BITS."
82 #endif
83 #define PTLRPC_BULK_OPS_COUNT   (1U << PTLRPC_BULK_OPS_BITS)
84 /**
85  * PTLRPC_BULK_OPS_MASK is for the convenience of the client only, and
86  * should not be used on the server at all.  Otherwise, it imposes a
87  * protocol limitation on the maximum RPC size that can be used by any
88  * RPC sent to that server in the future.  Instead, the server should
89  * use the negotiated per-client ocd_brw_size to determine the bulk
90  * RPC count. */
91 #define PTLRPC_BULK_OPS_MASK    (~((__u64)PTLRPC_BULK_OPS_COUNT - 1))
92
93 /**
94  * Define maxima for bulk I/O.
95  *
96  * A single PTLRPC BRW request is sent via up to PTLRPC_BULK_OPS_COUNT
97  * of LNET_MTU sized RDMA transfers.  Clients and servers negotiate the
98  * currently supported maximum between peers at connect via ocd_brw_size.
99  */
100 #define PTLRPC_MAX_BRW_BITS     (LNET_MTU_BITS + PTLRPC_BULK_OPS_BITS)
101 #define PTLRPC_MAX_BRW_SIZE     (1U << PTLRPC_MAX_BRW_BITS)
102 #define PTLRPC_MAX_BRW_PAGES    (PTLRPC_MAX_BRW_SIZE >> PAGE_SHIFT)
103
104 #define ONE_MB_BRW_SIZE         (1U << LNET_MTU_BITS)
105 #define MD_MAX_BRW_SIZE         (1U << LNET_MTU_BITS)
106 #define MD_MAX_BRW_PAGES        (MD_MAX_BRW_SIZE >> PAGE_SHIFT)
107 #define DT_MAX_BRW_SIZE         PTLRPC_MAX_BRW_SIZE
108 #define DT_DEF_BRW_SIZE         (4 * ONE_MB_BRW_SIZE)
109 #define DT_MAX_BRW_PAGES        (DT_MAX_BRW_SIZE >> PAGE_SHIFT)
110 #define OFD_MAX_BRW_SIZE        (1U << LNET_MTU_BITS)
111
112 /* When PAGE_SIZE is a constant, we can check our arithmetic here with cpp! */
113 #if ((PTLRPC_MAX_BRW_PAGES & (PTLRPC_MAX_BRW_PAGES - 1)) != 0)
114 # error "PTLRPC_MAX_BRW_PAGES isn't a power of two"
115 #endif
116 #if (PTLRPC_MAX_BRW_SIZE != (PTLRPC_MAX_BRW_PAGES * PAGE_SIZE))
117 # error "PTLRPC_MAX_BRW_SIZE isn't PTLRPC_MAX_BRW_PAGES * PAGE_SIZE"
118 #endif
119 #if (PTLRPC_MAX_BRW_SIZE > LNET_MTU * PTLRPC_BULK_OPS_COUNT)
120 # error "PTLRPC_MAX_BRW_SIZE too big"
121 #endif
122 #if (PTLRPC_MAX_BRW_PAGES > LNET_MAX_IOV * PTLRPC_BULK_OPS_COUNT)
123 # error "PTLRPC_MAX_BRW_PAGES too big"
124 #endif
125
126 #define PTLRPC_NTHRS_INIT       2
127
128 /**
129  * Buffer Constants
130  *
131  * Constants determine how memory is used to buffer incoming service requests.
132  *
133  * ?_NBUFS              # buffers to allocate when growing the pool
134  * ?_BUFSIZE            # bytes in a single request buffer
135  * ?_MAXREQSIZE         # maximum request service will receive
136  *
137  * When fewer than ?_NBUFS/2 buffers are posted for receive, another chunk
138  * of ?_NBUFS is added to the pool.
139  *
140  * Messages larger than ?_MAXREQSIZE are dropped.  Request buffers are
141  * considered full when less than ?_MAXREQSIZE is left in them.
142  */
143 /**
144  * Thread Constants
145  *
146  * Constants determine how threads are created for ptlrpc service.
147  *
148  * ?_NTHRS_INIT         # threads to create for each service partition on
149  *                        initializing. If it's non-affinity service and
150  *                        there is only one partition, it's the overall #
151  *                        threads for the service while initializing.
152  * ?_NTHRS_BASE         # threads should be created at least for each
153  *                        ptlrpc partition to keep the service healthy.
154  *                        It's the low-water mark of threads upper-limit
155  *                        for each partition.
156  * ?_THR_FACTOR         # threads can be added on threads upper-limit for
157  *                        each CPU core. This factor is only for reference,
158  *                        we might decrease value of factor if number of cores
159  *                        per CPT is above a limit.
160  * ?_NTHRS_MAX          # overall threads can be created for a service,
161  *                        it's a soft limit because if service is running
162  *                        on machine with hundreds of cores and tens of
163  *                        CPU partitions, we need to guarantee each partition
164  *                        has ?_NTHRS_BASE threads, which means total threads
165  *                        will be ?_NTHRS_BASE * number_of_cpts which can
166  *                        exceed ?_NTHRS_MAX.
167  *
168  * Examples
169  *
170  * #define MDS_NTHRS_INIT       2
171  * #define MDS_NTHRS_BASE       64
172  * #define MDS_NTHRS_FACTOR     8
173  * #define MDS_NTHRS_MAX        1024
174  *
175  * Example 1):
176  * ---------------------------------------------------------------------
177  * Server(A) has 16 cores, user configured it to 4 partitions so each
178  * partition has 4 cores, then actual number of service threads on each
179  * partition is:
180  *     MDS_NTHRS_BASE(64) + cores(4) * MDS_NTHRS_FACTOR(8) = 96
181  *
182  * Total number of threads for the service is:
183  *     96 * partitions(4) = 384
184  *
185  * Example 2):
186  * ---------------------------------------------------------------------
187  * Server(B) has 32 cores, user configured it to 4 partitions so each
188  * partition has 8 cores, then actual number of service threads on each
189  * partition is:
190  *     MDS_NTHRS_BASE(64) + cores(8) * MDS_NTHRS_FACTOR(8) = 128
191  *
192  * Total number of threads for the service is:
193  *     128 * partitions(4) = 512
194  *
195  * Example 3):
196  * ---------------------------------------------------------------------
197  * Server(B) has 96 cores, user configured it to 8 partitions so each
198  * partition has 12 cores, then actual number of service threads on each
199  * partition is:
200  *     MDS_NTHRS_BASE(64) + cores(12) * MDS_NTHRS_FACTOR(8) = 160
201  *
202  * Total number of threads for the service is:
203  *     160 * partitions(8) = 1280
204  *
205  * However, it's above the soft limit MDS_NTHRS_MAX, so we choose this number
206  * as upper limit of threads number for each partition:
207  *     MDS_NTHRS_MAX(1024) / partitions(8) = 128
208  *
209  * Example 4):
210  * ---------------------------------------------------------------------
211  * Server(C) have a thousand of cores and user configured it to 32 partitions
212  *     MDS_NTHRS_BASE(64) * 32 = 2048
213  *
214  * which is already above soft limit MDS_NTHRS_MAX(1024), but we still need
215  * to guarantee that each partition has at least MDS_NTHRS_BASE(64) threads
216  * to keep service healthy, so total number of threads will just be 2048.
217  *
218  * NB: we don't suggest to choose server with that many cores because backend
219  *     filesystem itself, buffer cache, or underlying network stack might
220  *     have some SMP scalability issues at that large scale.
221  *
222  *     If user already has a fat machine with hundreds or thousands of cores,
223  *     there are two choices for configuration:
224  *     a) create CPU table from subset of all CPUs and run Lustre on
225  *        top of this subset
226  *     b) bind service threads on a few partitions, see modparameters of
227  *        MDS and OSS for details
228 *
229  * NB: these calculations (and examples below) are simplified to help
230  *     understanding, the real implementation is a little more complex,
231  *     please see ptlrpc_server_nthreads_check() for details.
232  *
233  */
234
235  /*
236   * LDLM threads constants:
237   *
238   * Given 8 as factor and 24 as base threads number
239   *
240   * example 1)
241   * On 4-core machine we will have 24 + 8 * 4 = 56 threads.
242   *
243   * example 2)
244   * On 8-core machine with 2 partitions we will have 24 + 4 * 8 = 56
245   * threads for each partition and total threads number will be 112.
246   *
247   * example 3)
248   * On 64-core machine with 8 partitions we will need LDLM_NTHRS_BASE(24)
249   * threads for each partition to keep service healthy, so total threads
250   * number should be 24 * 8 = 192.
251   *
252   * So with these constants, threads number will be at the similar level
253   * of old versions, unless target machine has over a hundred cores
254   */
255 #define LDLM_THR_FACTOR         8
256 #define LDLM_NTHRS_INIT         PTLRPC_NTHRS_INIT
257 #define LDLM_NTHRS_BASE         24
258 #define LDLM_NTHRS_MAX          (num_online_cpus() == 1 ? 64 : 128)
259
260 #define LDLM_BL_THREADS   LDLM_NTHRS_AUTO_INIT
261 #define LDLM_CLIENT_NBUFS 1
262 #define LDLM_SERVER_NBUFS 64
263 #define LDLM_BUFSIZE      (8 * 1024)
264 #define LDLM_MAXREQSIZE   (5 * 1024)
265 #define LDLM_MAXREPSIZE   (1024)
266
267  /*
268   * MDS threads constants:
269   *
270   * Please see examples in "Thread Constants", MDS threads number will be at
271   * the comparable level of old versions, unless the server has many cores.
272   */
273 #ifndef MDS_MAX_THREADS
274 #define MDS_MAX_THREADS         1024
275 #define MDS_MAX_OTHR_THREADS    256
276
277 #else /* MDS_MAX_THREADS */
278 #if MDS_MAX_THREADS < PTLRPC_NTHRS_INIT
279 #undef MDS_MAX_THREADS
280 #define MDS_MAX_THREADS PTLRPC_NTHRS_INIT
281 #endif
282 #define MDS_MAX_OTHR_THREADS    max(PTLRPC_NTHRS_INIT, MDS_MAX_THREADS / 2)
283 #endif
284
285 /* default service */
286 #define MDS_THR_FACTOR          8
287 #define MDS_NTHRS_INIT          PTLRPC_NTHRS_INIT
288 #define MDS_NTHRS_MAX           MDS_MAX_THREADS
289 #define MDS_NTHRS_BASE          min(64, MDS_NTHRS_MAX)
290
291 /* read-page service */
292 #define MDS_RDPG_THR_FACTOR     4
293 #define MDS_RDPG_NTHRS_INIT     PTLRPC_NTHRS_INIT
294 #define MDS_RDPG_NTHRS_MAX      MDS_MAX_OTHR_THREADS
295 #define MDS_RDPG_NTHRS_BASE     min(48, MDS_RDPG_NTHRS_MAX)
296
297 /* these should be removed when we remove setattr service in the future */
298 #define MDS_SETA_THR_FACTOR     4
299 #define MDS_SETA_NTHRS_INIT     PTLRPC_NTHRS_INIT
300 #define MDS_SETA_NTHRS_MAX      MDS_MAX_OTHR_THREADS
301 #define MDS_SETA_NTHRS_BASE     min(48, MDS_SETA_NTHRS_MAX)
302
303 /* non-affinity threads */
304 #define MDS_OTHR_NTHRS_INIT     PTLRPC_NTHRS_INIT
305 #define MDS_OTHR_NTHRS_MAX      MDS_MAX_OTHR_THREADS
306
307 #define MDS_NBUFS               64
308
309 /**
310  * Assume file name length = FNAME_MAX = 256 (true for ext3).
311  *        path name length = PATH_MAX = 4096
312  *        LOV MD size max  = EA_MAX = 24 * 2000
313  *              (NB: 24 is size of lov_ost_data)
314  *        LOV LOGCOOKIE size max = 32 * 2000
315  *              (NB: 32 is size of llog_cookie)
316  * symlink:  FNAME_MAX + PATH_MAX  <- largest
317  * link:     FNAME_MAX + PATH_MAX  (mds_rec_link < mds_rec_create)
318  * rename:   FNAME_MAX + FNAME_MAX
319  * open:     FNAME_MAX + EA_MAX
320  *
321  * MDS_MAXREQSIZE ~= 4736 bytes =
322  * lustre_msg + ldlm_request + mdt_body + mds_rec_create + FNAME_MAX + PATH_MAX
323  * MDS_MAXREPSIZE ~= 8300 bytes = lustre_msg + llog_header
324  *
325  * Realistic size is about 512 bytes (20 character name + 128 char symlink),
326  * except in the open case where there are a large number of OSTs in a LOV.
327  */
328 #define MDS_MAXREQSIZE          (5 * 1024)      /* >= 4736 */
329 #define MDS_MAXREPSIZE          (9 * 1024)      /* >= 8300 */
330
331 /**
332  * MDS incoming request with LOV EA
333  * 24 = sizeof(struct lov_ost_data), i.e: replay of opencreate
334  */
335 #define MDS_LOV_MAXREQSIZE      max(MDS_MAXREQSIZE, \
336                                     362 + LOV_MAX_STRIPE_COUNT * 24)
337 /**
338  * MDS outgoing reply with LOV EA
339  *
340  * NB: max reply size Lustre 2.4+ client can get from old MDS is:
341  * LOV_MAX_STRIPE_COUNT * (llog_cookie + lov_ost_data) + extra bytes
342  *
343  * but 2.4 or later MDS will never send reply with llog_cookie to any
344  * version client. This macro is defined for server side reply buffer size.
345  */
346 #define MDS_LOV_MAXREPSIZE      MDS_LOV_MAXREQSIZE
347
348 /**
349  * This is the size of a maximum REINT_SETXATTR request:
350  *
351  *   lustre_msg          56 (32 + 4 x 5 + 4)
352  *   ptlrpc_body        184
353  *   mdt_rec_setxattr   136
354  *   lustre_capa        120
355  *   name               256 (XATTR_NAME_MAX)
356  *   value            65536 (XATTR_SIZE_MAX)
357  */
358 #define MDS_EA_MAXREQSIZE       66288
359
360 /**
361  * These are the maximum request and reply sizes (rounded up to 1 KB
362  * boundaries) for the "regular" MDS_REQUEST_PORTAL and MDS_REPLY_PORTAL.
363  */
364 #define MDS_REG_MAXREQSIZE      (((max(MDS_EA_MAXREQSIZE, \
365                                        MDS_LOV_MAXREQSIZE) + 1023) >> 10) << 10)
366 #define MDS_REG_MAXREPSIZE      MDS_REG_MAXREQSIZE
367
368 /**
369  * The update request includes all of updates from the create, which might
370  * include linkea (4K maxim), together with other updates, we set it to 1000K:
371  * lustre_msg + ptlrpc_body + OUT_UPDATE_BUFFER_SIZE_MAX
372  */
373 #define OUT_MAXREQSIZE  (1000 * 1024)
374 #define OUT_MAXREPSIZE  MDS_MAXREPSIZE
375
376 /** MDS_BUFSIZE = max_reqsize (w/o LOV EA) + max sptlrpc payload size */
377 #define MDS_BUFSIZE             max(MDS_MAXREQSIZE + SPTLRPC_MAX_PAYLOAD, \
378                                     8 * 1024)
379
380 /**
381  * MDS_REG_BUFSIZE should at least be MDS_REG_MAXREQSIZE + SPTLRPC_MAX_PAYLOAD.
382  * However, we need to allocate a much larger buffer for it because LNet
383  * requires each MD(rqbd) has at least MDS_REQ_MAXREQSIZE bytes left to avoid
384  * dropping of maximum-sized incoming request.  So if MDS_REG_BUFSIZE is only a
385  * little larger than MDS_REG_MAXREQSIZE, then it can only fit in one request
386  * even there are about MDS_REG_MAX_REQSIZE bytes left in a rqbd, and memory
387  * utilization is very low.
388  *
389  * In the meanwhile, size of rqbd can't be too large, because rqbd can't be
390  * reused until all requests fit in it have been processed and released,
391  * which means one long blocked request can prevent the rqbd be reused.
392  * Now we set request buffer size to 160 KB, so even each rqbd is unlinked
393  * from LNet with unused 65 KB, buffer utilization will be about 59%.
394  * Please check LU-2432 for details.
395  */
396 #define MDS_REG_BUFSIZE         max(MDS_REG_MAXREQSIZE + SPTLRPC_MAX_PAYLOAD, \
397                                     160 * 1024)
398
399 /**
400  * OUT_BUFSIZE = max_out_reqsize + max sptlrpc payload (~1K) which is
401  * about 10K, for the same reason as MDS_REG_BUFSIZE, we also give some
402  * extra bytes to each request buffer to improve buffer utilization rate.
403   */
404 #define OUT_BUFSIZE             max(OUT_MAXREQSIZE + SPTLRPC_MAX_PAYLOAD, \
405                                     24 * 1024)
406
407 /** FLD_MAXREQSIZE == lustre_msg + __u32 padding + ptlrpc_body + opc */
408 #define FLD_MAXREQSIZE  (160)
409
410 /** FLD_MAXREPSIZE == lustre_msg + ptlrpc_body */
411 #define FLD_MAXREPSIZE  (152)
412 #define FLD_BUFSIZE     (1 << 12)
413
414 /**
415  * SEQ_MAXREQSIZE == lustre_msg + __u32 padding + ptlrpc_body + opc + lu_range +
416  * __u32 padding */
417 #define SEQ_MAXREQSIZE  (160)
418
419 /** SEQ_MAXREPSIZE == lustre_msg + ptlrpc_body + lu_range */
420 #define SEQ_MAXREPSIZE  (152)
421 #define SEQ_BUFSIZE     (1 << 12)
422
423 /** MGS threads must be >= 3, see bug 22458 comment #28 */
424 #define MGS_NTHRS_INIT  (PTLRPC_NTHRS_INIT + 1)
425 #define MGS_NTHRS_MAX   32
426
427 #define MGS_NBUFS       64
428 #define MGS_BUFSIZE     (8 * 1024)
429 #define MGS_MAXREQSIZE  (7 * 1024)
430 #define MGS_MAXREPSIZE  (9 * 1024)
431
432  /*
433   * OSS threads constants:
434   *
435   * Given 8 as factor and 64 as base threads number
436   *
437   * example 1):
438   * On 8-core server configured to 2 partitions, we will have
439   * 64 + 8 * 4 = 96 threads for each partition, 192 total threads.
440   *
441   * example 2):
442   * On 32-core machine configured to 4 partitions, we will have
443   * 64 + 8 * 8 = 112 threads for each partition, so total threads number
444   * will be 112 * 4 = 448.
445   *
446   * example 3):
447   * On 64-core machine configured to 4 partitions, we will have
448   * 64 + 16 * 8 = 192 threads for each partition, so total threads number
449   * will be 192 * 4 = 768 which is above limit OSS_NTHRS_MAX(512), so we
450   * cut off the value to OSS_NTHRS_MAX(512) / 4 which is 128 threads
451   * for each partition.
452   *
453   * So we can see that with these constants, threads number wil be at the
454   * similar level of old versions, unless the server has many cores.
455   */
456  /* depress threads factor for VM with small memory size */
457 #define OSS_THR_FACTOR          min_t(int, 8, \
458                                 NUM_CACHEPAGES >> (28 - PAGE_SHIFT))
459 #define OSS_NTHRS_INIT          (PTLRPC_NTHRS_INIT + 1)
460 #define OSS_NTHRS_BASE          64
461
462 /* threads for handling "create" request */
463 #define OSS_CR_THR_FACTOR       1
464 #define OSS_CR_NTHRS_INIT       PTLRPC_NTHRS_INIT
465 #define OSS_CR_NTHRS_BASE       8
466 #define OSS_CR_NTHRS_MAX        64
467
468 /**
469  * OST_IO_MAXREQSIZE ~=
470  *      lustre_msg + ptlrpc_body + obdo + obd_ioobj +
471  *      DT_MAX_BRW_PAGES * niobuf_remote
472  *
473  * - single object with 16 pages is 512 bytes
474  * - OST_IO_MAXREQSIZE must be at least 1 niobuf per page of data
475  * - Must be a multiple of 1024
476  * - should allow a reasonably large SHORT_IO_BYTES size (64KB)
477  */
478 #define _OST_MAXREQSIZE_BASE ((unsigned long)(sizeof(struct lustre_msg)   + \
479                              /* lm_buflens */ sizeof(__u32) * 4           + \
480                                               sizeof(struct ptlrpc_body)  + \
481                                               sizeof(struct obdo)         + \
482                                               sizeof(struct obd_ioobj)    + \
483                                               sizeof(struct niobuf_remote)))
484 #define _OST_MAXREQSIZE_SUM ((unsigned long)(_OST_MAXREQSIZE_BASE         + \
485                                              sizeof(struct niobuf_remote) * \
486                                              DT_MAX_BRW_PAGES))
487 /**
488  * FIEMAP request can be 4K+ for now
489  */
490 #define OST_MAXREQSIZE          (16UL * 1024UL)
491 #define OST_IO_MAXREQSIZE       max(OST_MAXREQSIZE,                     \
492                                    ((_OST_MAXREQSIZE_SUM - 1) |         \
493                                     (1024UL - 1)) + 1)
494 /* Safe estimate of free space in standard RPC, provides upper limit for # of
495  * bytes of i/o to pack in RPC (skipping bulk transfer). */
496 #define OST_MAX_SHORT_IO_BYTES  ((OST_IO_MAXREQSIZE - _OST_MAXREQSIZE_BASE) & \
497                                  PAGE_MASK)
498
499 /* Actual size used for short i/o buffer.  Calculation means this:
500  * At least one page (for large PAGE_SIZE), or 16 KiB, but not more
501  * than the available space aligned to a page boundary. */
502 #define OBD_DEF_SHORT_IO_BYTES  min(max(PAGE_SIZE, 16UL * 1024UL), \
503                                     OST_MAX_SHORT_IO_BYTES)
504
505 #define OST_MAXREPSIZE          (9 * 1024)
506 #define OST_IO_MAXREPSIZE       OST_MAXREPSIZE
507
508 #define OST_NBUFS               64
509 /** OST_BUFSIZE = max_reqsize + max sptlrpc payload size */
510 #define OST_BUFSIZE             max_t(int, OST_MAXREQSIZE + 1024, 32 * 1024)
511 /**
512  * OST_IO_MAXREQSIZE is 18K, giving extra 46K can increase buffer utilization
513  * rate of request buffer, please check comment of MDS_LOV_BUFSIZE for details.
514  */
515 #define OST_IO_BUFSIZE          max_t(int, OST_IO_MAXREQSIZE + 1024, 64 * 1024)
516
517 /* Macro to hide a typecast and BUILD_BUG. */
518 #define ptlrpc_req_async_args(_var, req) ({                             \
519                 BUILD_BUG_ON(sizeof(*_var) > sizeof(req->rq_async_args)); \
520                 (typeof(_var))&req->rq_async_args;                      \
521         })
522
523 struct ptlrpc_replay_async_args {
524         int             praa_old_state;
525         int             praa_old_status;
526 };
527
528 /**
529  * Structure to single define portal connection.
530  */
531 struct ptlrpc_connection {
532         /** linkage for connections hash table */
533         struct hlist_node        c_hash;
534         /** Our own lnet nid for this connection */
535         lnet_nid_t              c_self;
536         /** Remote side nid for this connection */
537         struct lnet_process_id       c_peer;
538         /** UUID of the other side */
539         struct obd_uuid         c_remote_uuid;
540         /** reference counter for this connection */
541         atomic_t            c_refcount;
542 };
543
544 /** Client definition for PortalRPC */
545 struct ptlrpc_client {
546         /** What lnet portal does this client send messages to by default */
547         __u32                   cli_request_portal;
548         /** What portal do we expect replies on */
549         __u32                   cli_reply_portal;
550         /** Name of the client */
551         const char              *cli_name;
552 };
553
554 /** state flags of requests */
555 /* XXX only ones left are those used by the bulk descs as well! */
556 #define PTL_RPC_FL_INTR         BIT(0)  /* reply wait was interrupted by user */
557 #define PTL_RPC_FL_TIMEOUT      BIT(7)  /* request timed out waiting for reply */
558
559 #define REQ_MAX_ACK_LOCKS 8
560
561 union ptlrpc_async_args {
562         /**
563          * Scratchpad for passing args to completion interpreter. Users
564          * cast to the struct of their choosing, and BUILD_BUG_ON that this is
565          * big enough.  For _tons_ of context, OBD_ALLOC a struct and store
566          * a pointer to it here.  The pointer_arg ensures this struct is at
567          * least big enough for that.
568          */
569         void    *pointer_arg[11];
570         __u64   space[7];
571 };
572
573 struct ptlrpc_request_set;
574 typedef int (*set_producer_func)(struct ptlrpc_request_set *, void *);
575
576 /**
577  * Definition of request set structure.
578  * Request set is a list of requests (not necessary to the same target) that
579  * once populated with RPCs could be sent in parallel.
580  * There are two kinds of request sets. General purpose and with dedicated
581  * serving thread. Example of the latter is ptlrpcd set.
582  * For general purpose sets once request set started sending it is impossible
583  * to add new requests to such set.
584  * Provides a way to call "completion callbacks" when all requests in the set
585  * returned.
586  */
587 struct ptlrpc_request_set {
588         atomic_t                set_refcount;
589         /** number of in queue requests */
590         atomic_t                set_new_count;
591         /** number of uncompleted requests */
592         atomic_t                set_remaining;
593         /** wait queue to wait on for request events */
594         wait_queue_head_t       set_waitq;
595         /** List of requests in the set */
596         struct list_head        set_requests;
597         /**
598          * Lock for \a set_new_requests manipulations
599          * locked so that any old caller can communicate requests to
600          * the set holder who can then fold them into the lock-free set
601          */
602         spinlock_t              set_new_req_lock;
603         /** List of new yet unsent requests. Only used with ptlrpcd now. */
604         struct list_head        set_new_requests;
605
606         /** rq_status of requests that have been freed already */
607         int                     set_rc;
608         /** Additional fields used by the flow control extension */
609         /** Maximum number of RPCs in flight */
610         int                     set_max_inflight;
611         /** Callback function used to generate RPCs */
612         set_producer_func       set_producer;
613         /** opaq argument passed to the producer callback */
614         void                    *set_producer_arg;
615         unsigned int             set_allow_intr:1;
616 };
617
618 struct ptlrpc_bulk_desc;
619 struct ptlrpc_service_part;
620 struct ptlrpc_service;
621
622 /**
623  * ptlrpc callback & work item stuff
624  */
625 struct ptlrpc_cb_id {
626         void (*cbid_fn)(struct lnet_event *ev); /* specific callback fn */
627         void *cbid_arg;                         /* additional arg */
628 };
629
630 /** Maximum number of locks to fit into reply state */
631 #define RS_MAX_LOCKS 8
632 #define RS_DEBUG     0
633
634 /**
635  * Structure to define reply state on the server
636  * Reply state holds various reply message information. Also for "difficult"
637  * replies (rep-ack case) we store the state after sending reply and wait
638  * for the client to acknowledge the reception. In these cases locks could be
639  * added to the state for replay/failover consistency guarantees.
640  */
641 struct ptlrpc_reply_state {
642         /** Callback description */
643         struct ptlrpc_cb_id     rs_cb_id;
644         /** Linkage for list of all reply states in a system */
645         struct list_head        rs_list;
646         /** Linkage for list of all reply states on same export */
647         struct list_head        rs_exp_list;
648         /** Linkage for list of all reply states for same obd */
649         struct list_head        rs_obd_list;
650 #if RS_DEBUG
651         struct list_head        rs_debug_list;
652 #endif
653         /** A spinlock to protect the reply state flags */
654         spinlock_t              rs_lock;
655         /** Reply state flags */
656         unsigned long          rs_difficult:1;     /* ACK/commit stuff */
657         unsigned long          rs_no_ack:1;    /* no ACK, even for
658                                                   difficult requests */
659         unsigned long          rs_scheduled:1;     /* being handled? */
660         unsigned long          rs_scheduled_ever:1;/* any schedule attempts? */
661         unsigned long          rs_handled:1;  /* been handled yet? */
662         unsigned long          rs_on_net:1;   /* reply_out_callback pending? */
663         unsigned long          rs_prealloc:1; /* rs from prealloc list */
664         unsigned long          rs_committed:1;/* the transaction was committed
665                                                  and the rs was dispatched
666                                                  by ptlrpc_commit_replies */
667         unsigned long           rs_convert_lock:1; /* need to convert saved
668                                                     * locks to COS mode */
669         atomic_t                rs_refcount;    /* number of users */
670         /** Number of locks awaiting client ACK */
671         int                     rs_nlocks;
672
673         /** Size of the state */
674         int                    rs_size;
675         /** opcode */
676         __u32                  rs_opc;
677         /** Transaction number */
678         __u64                  rs_transno;
679         /** xid */
680         __u64                  rs_xid;
681         struct obd_export     *rs_export;
682         struct ptlrpc_service_part *rs_svcpt;
683         /** Lnet metadata handle for the reply */
684         struct lnet_handle_md   rs_md_h;
685
686         /** Context for the sevice thread */
687         struct ptlrpc_svc_ctx   *rs_svc_ctx;
688         /** Reply buffer (actually sent to the client), encoded if needed */
689         struct lustre_msg       *rs_repbuf;     /* wrapper */
690         /** Size of the reply buffer */
691         int                     rs_repbuf_len;  /* wrapper buf length */
692         /** Size of the reply message */
693         int                     rs_repdata_len; /* wrapper msg length */
694         /**
695          * Actual reply message. Its content is encrupted (if needed) to
696          * produce reply buffer for actual sending. In simple case
697          * of no network encryption we jus set \a rs_repbuf to \a rs_msg
698          */
699         struct lustre_msg       *rs_msg;        /* reply message */
700
701         /** Handles of locks awaiting client reply ACK */
702         struct lustre_handle    rs_locks[RS_MAX_LOCKS];
703         /** Lock modes of locks in \a rs_locks */
704         enum ldlm_mode          rs_modes[RS_MAX_LOCKS];
705 };
706
707 struct ptlrpc_thread;
708
709 /** RPC stages */
710 enum rq_phase {
711         RQ_PHASE_NEW            = 0xebc0de00,
712         RQ_PHASE_RPC            = 0xebc0de01,
713         RQ_PHASE_BULK           = 0xebc0de02,
714         RQ_PHASE_INTERPRET      = 0xebc0de03,
715         RQ_PHASE_COMPLETE       = 0xebc0de04,
716         RQ_PHASE_UNREG_RPC      = 0xebc0de05,
717         RQ_PHASE_UNREG_BULK     = 0xebc0de06,
718         RQ_PHASE_UNDEFINED      = 0xebc0de07
719 };
720
721 /** Type of request interpreter call-back */
722 typedef int (*ptlrpc_interpterer_t)(const struct lu_env *env,
723                                     struct ptlrpc_request *req,
724                                     void *arg, int rc);
725 /** Type of request resend call-back */
726 typedef void (*ptlrpc_resend_cb_t)(struct ptlrpc_request *req,
727                                    void *arg);
728
729 /**
730  * Definition of request pool structure.
731  * The pool is used to store empty preallocated requests for the case
732  * when we would actually need to send something without performing
733  * any allocations (to avoid e.g. OOM).
734  */
735 struct ptlrpc_request_pool {
736         /** Locks the list */
737         spinlock_t              prp_lock;
738         /** list of ptlrpc_request structs */
739         struct list_head        prp_req_list;
740         /** Maximum message size that would fit into a rquest from this pool */
741         int                     prp_rq_size;
742         /** Function to allocate more requests for this pool */
743         int (*prp_populate)(struct ptlrpc_request_pool *, int);
744 };
745
746 struct lu_context;
747 struct lu_env;
748
749 struct ldlm_lock;
750
751 #include <lustre_nrs.h>
752
753 /**
754  * Basic request prioritization operations structure.
755  * The whole idea is centered around locks and RPCs that might affect locks.
756  * When a lock is contended we try to give priority to RPCs that might lead
757  * to fastest release of that lock.
758  * Currently only implemented for OSTs only in a way that makes all
759  * IO and truncate RPCs that are coming from a locked region where a lock is
760  * contended a priority over other requests.
761  */
762 struct ptlrpc_hpreq_ops {
763         /**
764          * Check if the lock handle of the given lock is the same as
765          * taken from the request.
766          */
767         int  (*hpreq_lock_match)(struct ptlrpc_request *, struct ldlm_lock *);
768         /**
769          * Check if the request is a high priority one.
770          */
771         int  (*hpreq_check)(struct ptlrpc_request *);
772         /**
773          * Called after the request has been handled.
774          */
775         void (*hpreq_fini)(struct ptlrpc_request *);
776 };
777
778 struct ptlrpc_cli_req {
779         /** For bulk requests on client only: bulk descriptor */
780         struct ptlrpc_bulk_desc         *cr_bulk;
781         /** optional time limit for send attempts. This is a timeout
782          *  not a timestamp so timeout_t (s32) is used instead of time64_t
783          */
784         timeout_t                        cr_delay_limit;
785         /** time request was first queued */
786         time64_t                         cr_queued_time;
787         /** request sent in nanoseconds */
788         ktime_t                          cr_sent_ns;
789         /** time for request really sent out */
790         time64_t                         cr_sent_out;
791         /** when req reply unlink must finish. */
792         time64_t                         cr_reply_deadline;
793         /** when req bulk unlink must finish. */
794         time64_t                         cr_bulk_deadline;
795         /** when req unlink must finish. */
796         time64_t                         cr_req_deadline;
797         /** Portal to which this request would be sent */
798         short                            cr_req_ptl;
799         /** Portal where to wait for reply and where reply would be sent */
800         short                            cr_rep_ptl;
801         /** request resending number */
802         unsigned int                     cr_resend_nr;
803         /** What was import generation when this request was sent */
804         int                              cr_imp_gen;
805         enum lustre_imp_state            cr_send_state;
806         /** Per-request waitq introduced by bug 21938 for recovery waiting */
807         wait_queue_head_t                cr_set_waitq;
808         /** Link item for request set lists */
809         struct list_head                 cr_set_chain;
810         /** link to waited ctx */
811         struct list_head                 cr_ctx_chain;
812
813         /** client's half ctx */
814         struct ptlrpc_cli_ctx           *cr_cli_ctx;
815         /** Link back to the request set */
816         struct ptlrpc_request_set       *cr_set;
817         /** outgoing request MD handle */
818         struct lnet_handle_md            cr_req_md_h;
819         /** request-out callback parameter */
820         struct ptlrpc_cb_id              cr_req_cbid;
821         /** incoming reply MD handle */
822         struct lnet_handle_md            cr_reply_md_h;
823         wait_queue_head_t                cr_reply_waitq;
824         /** reply callback parameter */
825         struct ptlrpc_cb_id              cr_reply_cbid;
826         /** Async completion handler, called when reply is received */
827         ptlrpc_interpterer_t             cr_reply_interp;
828         /** Resend handler, called when request is resend to update RPC data */
829         ptlrpc_resend_cb_t               cr_resend_cb;
830         /** Async completion context */
831         union ptlrpc_async_args          cr_async_args;
832         /** Opaq data for replay and commit callbacks. */
833         void                            *cr_cb_data;
834         /** Link to the imp->imp_unreplied_list */
835         struct list_head                 cr_unreplied_list;
836         /**
837          * Commit callback, called when request is committed and about to be
838          * freed.
839          */
840         void (*cr_commit_cb)(struct ptlrpc_request *);
841         /** Replay callback, called after request is replayed at recovery */
842         void (*cr_replay_cb)(struct ptlrpc_request *);
843 };
844
845 /** client request member alias */
846 /* NB: these alias should NOT be used by any new code, instead they should
847  * be removed step by step to avoid potential abuse */
848 #define rq_bulk                 rq_cli.cr_bulk
849 #define rq_delay_limit          rq_cli.cr_delay_limit
850 #define rq_queued_time          rq_cli.cr_queued_time
851 #define rq_sent_ns              rq_cli.cr_sent_ns
852 #define rq_real_sent            rq_cli.cr_sent_out
853 #define rq_reply_deadline       rq_cli.cr_reply_deadline
854 #define rq_bulk_deadline        rq_cli.cr_bulk_deadline
855 #define rq_req_deadline         rq_cli.cr_req_deadline
856 #define rq_nr_resend            rq_cli.cr_resend_nr
857 #define rq_request_portal       rq_cli.cr_req_ptl
858 #define rq_reply_portal         rq_cli.cr_rep_ptl
859 #define rq_import_generation    rq_cli.cr_imp_gen
860 #define rq_send_state           rq_cli.cr_send_state
861 #define rq_set_chain            rq_cli.cr_set_chain
862 #define rq_ctx_chain            rq_cli.cr_ctx_chain
863 #define rq_set                  rq_cli.cr_set
864 #define rq_set_waitq            rq_cli.cr_set_waitq
865 #define rq_cli_ctx              rq_cli.cr_cli_ctx
866 #define rq_req_md_h             rq_cli.cr_req_md_h
867 #define rq_req_cbid             rq_cli.cr_req_cbid
868 #define rq_reply_md_h           rq_cli.cr_reply_md_h
869 #define rq_reply_waitq          rq_cli.cr_reply_waitq
870 #define rq_reply_cbid           rq_cli.cr_reply_cbid
871 #define rq_interpret_reply      rq_cli.cr_reply_interp
872 #define rq_resend_cb            rq_cli.cr_resend_cb
873 #define rq_async_args           rq_cli.cr_async_args
874 #define rq_cb_data              rq_cli.cr_cb_data
875 #define rq_unreplied_list       rq_cli.cr_unreplied_list
876 #define rq_commit_cb            rq_cli.cr_commit_cb
877 #define rq_replay_cb            rq_cli.cr_replay_cb
878
879 struct ptlrpc_srv_req {
880         /** initial thread servicing this request */
881         struct ptlrpc_thread            *sr_svc_thread;
882         /**
883          * Server side list of incoming unserved requests sorted by arrival
884          * time.  Traversed from time to time to notice about to expire
885          * requests and sent back "early replies" to clients to let them
886          * know server is alive and well, just very busy to service their
887          * requests in time
888          */
889         struct list_head                 sr_timed_list;
890         /** server-side per-export list */
891         struct list_head                 sr_exp_list;
892         /** server-side history, used for debuging purposes. */
893         struct list_head                 sr_hist_list;
894         /** history sequence # */
895         __u64                            sr_hist_seq;
896         /** the index of service's srv_at_array into which request is linked */
897         __u32                            sr_at_index;
898         /** authed uid */
899         uid_t                            sr_auth_uid;
900         /** authed uid mapped to */
901         uid_t                            sr_auth_mapped_uid;
902         /** RPC is generated from what part of Lustre */
903         enum lustre_sec_part             sr_sp_from;
904         /** request session context */
905         struct lu_context                sr_ses;
906         /** \addtogroup  nrs
907          * @{
908          */
909         /** stub for NRS request */
910         struct ptlrpc_nrs_request        sr_nrq;
911         /** @} nrs */
912         /** request arrival time */
913         struct timespec64                sr_arrival_time;
914         /** server's half ctx */
915         struct ptlrpc_svc_ctx           *sr_svc_ctx;
916         /** (server side), pointed directly into req buffer */
917         struct ptlrpc_user_desc         *sr_user_desc;
918         /** separated reply state, may be vmalloc'd */
919         struct ptlrpc_reply_state       *sr_reply_state;
920         /** server-side hp handlers */
921         struct ptlrpc_hpreq_ops         *sr_ops;
922         /** incoming request buffer */
923         struct ptlrpc_request_buffer_desc *sr_rqbd;
924 };
925
926 /** server request member alias */
927 /* NB: these alias should NOT be used by any new code, instead they should
928  * be removed step by step to avoid potential abuse */
929 #define rq_svc_thread           rq_srv.sr_svc_thread
930 #define rq_timed_list           rq_srv.sr_timed_list
931 #define rq_exp_list             rq_srv.sr_exp_list
932 #define rq_history_list         rq_srv.sr_hist_list
933 #define rq_history_seq          rq_srv.sr_hist_seq
934 #define rq_at_index             rq_srv.sr_at_index
935 #define rq_auth_uid             rq_srv.sr_auth_uid
936 #define rq_auth_mapped_uid      rq_srv.sr_auth_mapped_uid
937 #define rq_sp_from              rq_srv.sr_sp_from
938 #define rq_session              rq_srv.sr_ses
939 #define rq_nrq                  rq_srv.sr_nrq
940 #define rq_arrival_time         rq_srv.sr_arrival_time
941 #define rq_reply_state          rq_srv.sr_reply_state
942 #define rq_svc_ctx              rq_srv.sr_svc_ctx
943 #define rq_user_desc            rq_srv.sr_user_desc
944 #define rq_ops                  rq_srv.sr_ops
945 #define rq_rqbd                 rq_srv.sr_rqbd
946
947 /**
948  * Represents remote procedure call.
949  *
950  * This is a staple structure used by everybody wanting to send a request
951  * in Lustre.
952  */
953 struct ptlrpc_request {
954         /* Request type: one of PTL_RPC_MSG_* */
955         int                              rq_type;
956         /** Result of request processing */
957         int                              rq_status;
958         /**
959          * Linkage item through which this request is included into
960          * sending/delayed lists on client and into rqbd list on server
961          */
962         struct list_head                 rq_list;
963         /** Lock to protect request flags and some other important bits, like
964          * rq_list
965          */
966         spinlock_t                       rq_lock;
967         spinlock_t                       rq_early_free_lock;
968         /** client-side flags are serialized by rq_lock @{ */
969         unsigned int rq_intr:1, rq_replied:1, rq_err:1,
970                 rq_timedout:1, rq_resend:1, rq_restart:1,
971                 /**
972                  * when ->rq_replay is set, request is kept by the client even
973                  * after server commits corresponding transaction. This is
974                  * used for operations that require sequence of multiple
975                  * requests to be replayed. The only example currently is file
976                  * open/close. When last request in such a sequence is
977                  * committed, ->rq_replay is cleared on all requests in the
978                  * sequence.
979                  */
980                 rq_replay:1,
981                 rq_no_resend:1, rq_waiting:1, rq_receiving_reply:1,
982                 rq_no_delay:1, rq_net_err:1, rq_wait_ctx:1,
983                 rq_early:1,
984                 rq_req_unlinked:1,      /* unlinked request buffer from lnet */
985                 rq_reply_unlinked:1,    /* unlinked reply buffer from lnet */
986                 rq_memalloc:1,      /* req originated from "kswapd" */
987                 rq_committed:1,
988                 rq_reply_truncated:1,
989                 /** whether the "rq_set" is a valid one */
990                 rq_invalid_rqset:1,
991                 rq_generation_set:1,
992                 /** do not resend request on -EINPROGRESS */
993                 rq_no_retry_einprogress:1,
994                 /* allow the req to be sent if the import is in recovery
995                  * status */
996                 rq_allow_replay:1,
997                 /* bulk request, sent to server, but uncommitted */
998                 rq_unstable:1,
999                 rq_early_free_repbuf:1, /* free reply buffer in advance */
1000                 rq_allow_intr:1;
1001         /** @} */
1002
1003         /** server-side flags @{ */
1004         unsigned int
1005                 rq_hp:1,                /**< high priority RPC */
1006                 rq_at_linked:1,         /**< link into service's srv_at_array */
1007                 rq_packed_final:1,      /**< packed final reply */
1008                 rq_obsolete:1;          /* aborted by a signal on a client */
1009         /** @} */
1010
1011         /** one of RQ_PHASE_* */
1012         enum rq_phase                    rq_phase;
1013         /** one of RQ_PHASE_* to be used next */
1014         enum rq_phase                    rq_next_phase;
1015         /**
1016          * client-side refcount for SENT race, server-side refcounf
1017          * for multiple replies
1018          */
1019         atomic_t                         rq_refcount;
1020         /**
1021          * client-side:
1022          * !rq_truncate : # reply bytes actually received,
1023          *  rq_truncate : required repbuf_len for resend
1024          */
1025         int rq_nob_received;
1026         /** Request length */
1027         int rq_reqlen;
1028         /** Reply length */
1029         int rq_replen;
1030         /** Pool if request is from preallocated list */
1031         struct ptlrpc_request_pool      *rq_pool;
1032         /** Request message - what client sent */
1033         struct lustre_msg *rq_reqmsg;
1034         /** Reply message - server response */
1035         struct lustre_msg *rq_repmsg;
1036         /** Transaction number */
1037         __u64 rq_transno;
1038         /** xid */
1039         __u64                            rq_xid;
1040         /** bulk match bits */
1041         __u64                            rq_mbits;
1042         /**
1043          * List item to for replay list. Not yet committed requests get linked
1044          * there.
1045          * Also see \a rq_replay comment above.
1046          * It's also link chain on obd_export::exp_req_replay_queue
1047          */
1048         struct list_head                 rq_replay_list;
1049         /** non-shared members for client & server request*/
1050         union {
1051                 struct ptlrpc_cli_req    rq_cli;
1052                 struct ptlrpc_srv_req    rq_srv;
1053         };
1054         /**
1055          * security and encryption data
1056          * @{ */
1057         /** description of flavors for client & server */
1058         struct sptlrpc_flavor            rq_flvr;
1059
1060         /**
1061          * SELinux policy info at the time of the request
1062          * sepol string format is:
1063          * <mode>:<policy name>:<policy version>:<policy hash>
1064          */
1065         char rq_sepol[LUSTRE_NODEMAP_SEPOL_LENGTH + 1];
1066
1067         /* client/server security flags */
1068         unsigned int
1069                                  rq_ctx_init:1,      /* context initiation */
1070                                  rq_ctx_fini:1,      /* context destroy */
1071                                  rq_bulk_read:1,     /* request bulk read */
1072                                  rq_bulk_write:1,    /* request bulk write */
1073                                  /* server authentication flags */
1074                                  rq_auth_gss:1,      /* authenticated by gss */
1075                                  rq_auth_usr_root:1, /* authed as root */
1076                                  rq_auth_usr_mdt:1,  /* authed as mdt */
1077                                  rq_auth_usr_ost:1,  /* authed as ost */
1078                                  /* security tfm flags */
1079                                  rq_pack_udesc:1,
1080                                  rq_pack_bulk:1,
1081                                  /* doesn't expect reply FIXME */
1082                                  rq_no_reply:1,
1083                                  rq_pill_init:1, /* pill initialized */
1084                                  rq_srv_req:1; /* server request */
1085
1086
1087         /** various buffer pointers */
1088         struct lustre_msg               *rq_reqbuf;  /**< req wrapper, vmalloc*/
1089         char                            *rq_repbuf;  /**< rep buffer, vmalloc */
1090         struct lustre_msg               *rq_repdata; /**< rep wrapper msg */
1091         /** only in priv mode */
1092         struct lustre_msg               *rq_clrbuf;
1093         int                      rq_reqbuf_len;  /* req wrapper buf len */
1094         int                      rq_reqdata_len; /* req wrapper msg len */
1095         int                      rq_repbuf_len;  /* rep buffer len */
1096         int                      rq_repdata_len; /* rep wrapper msg len */
1097         int                      rq_clrbuf_len;  /* only in priv mode */
1098         int                      rq_clrdata_len; /* only in priv mode */
1099
1100         /** early replies go to offset 0, regular replies go after that */
1101         unsigned int                     rq_reply_off;
1102         /** @} */
1103
1104         /** Fields that help to see if request and reply were swabbed or not */
1105         __u32                            rq_req_swab_mask;
1106         __u32                            rq_rep_swab_mask;
1107
1108         /** how many early replies (for stats) */
1109         int                              rq_early_count;
1110         /** Server-side, export on which request was received */
1111         struct obd_export               *rq_export;
1112         /** import where request is being sent */
1113         struct obd_import               *rq_import;
1114         /** our LNet NID */
1115         lnet_nid_t                       rq_self;
1116         /** Peer description (the other side) */
1117         struct lnet_process_id           rq_peer;
1118         /** Descriptor for the NID from which the peer sent the request. */
1119         struct lnet_process_id           rq_source;
1120         /**
1121          * service time estimate (secs)
1122          * If the request is not served by this time, it is marked as timed out.
1123          * Do not change to time64_t since this is transmitted over the wire.
1124          *
1125          * The linux kernel handles timestamps with time64_t and timeouts
1126          * are normally done with jiffies. Lustre shares the rq_timeout between
1127          * nodes. Since jiffies can vary from node to node Lustre instead
1128          * will express the timeout value in seconds. To avoid confusion with
1129          * timestamps (time64_t) and jiffy timeouts (long) Lustre timeouts
1130          * are expressed in s32 (timeout_t). Also what is transmitted over
1131          * the wire is 32 bits.
1132          */
1133         timeout_t                        rq_timeout;
1134         /**
1135          * when request/reply sent (secs), or time when request should be sent
1136          */
1137         time64_t                         rq_sent;
1138         /** when request must finish. */
1139         time64_t                         rq_deadline;
1140         /** request format description */
1141         struct req_capsule               rq_pill;
1142 };
1143
1144 /**
1145  * Call completion handler for rpc if any, return it's status or original
1146  * rc if there was no handler defined for this request.
1147  */
1148 static inline int ptlrpc_req_interpret(const struct lu_env *env,
1149                                        struct ptlrpc_request *req, int rc)
1150 {
1151         if (req->rq_interpret_reply != NULL) {
1152                 req->rq_status = req->rq_interpret_reply(env, req,
1153                                                          &req->rq_async_args,
1154                                                          rc);
1155                 return req->rq_status;
1156         }
1157
1158         return rc;
1159 }
1160
1161 /** \addtogroup  nrs
1162  * @{
1163  */
1164 int ptlrpc_nrs_policy_register(struct ptlrpc_nrs_pol_conf *conf);
1165 int ptlrpc_nrs_policy_unregister(struct ptlrpc_nrs_pol_conf *conf);
1166 void ptlrpc_nrs_req_hp_move(struct ptlrpc_request *req);
1167 void nrs_policy_get_info_locked(struct ptlrpc_nrs_policy *policy,
1168                                 struct ptlrpc_nrs_pol_info *info);
1169
1170 /*
1171  * Can the request be moved from the regular NRS head to the high-priority NRS
1172  * head (of the same PTLRPC service partition), if any?
1173  *
1174  * For a reliable result, this should be checked under svcpt->scp_req lock.
1175  */
1176 static inline bool ptlrpc_nrs_req_can_move(struct ptlrpc_request *req)
1177 {
1178         struct ptlrpc_nrs_request *nrq = &req->rq_nrq;
1179
1180         /**
1181          * LU-898: Check ptlrpc_nrs_request::nr_enqueued to make sure the
1182          * request has been enqueued first, and ptlrpc_nrs_request::nr_started
1183          * to make sure it has not been scheduled yet (analogous to previous
1184          * (non-NRS) checking of !list_empty(&ptlrpc_request::rq_list).
1185          */
1186         return nrq->nr_enqueued && !nrq->nr_started && !req->rq_hp;
1187 }
1188 /** @} nrs */
1189
1190 /**
1191  * Returns true if request buffer at offset \a index was already swabbed
1192  */
1193 static inline bool lustre_req_swabbed(struct ptlrpc_request *req, size_t index)
1194 {
1195         LASSERT(index < sizeof(req->rq_req_swab_mask) * 8);
1196         return req->rq_req_swab_mask & BIT(index);
1197 }
1198
1199 /**
1200  * Returns true if request reply buffer at offset \a index was already swabbed
1201  */
1202 static inline bool lustre_rep_swabbed(struct ptlrpc_request *req, size_t index)
1203 {
1204         LASSERT(index < sizeof(req->rq_rep_swab_mask) * 8);
1205         return req->rq_rep_swab_mask & BIT(index);
1206 }
1207
1208 /**
1209  * Returns true if request needs to be swabbed into local cpu byteorder
1210  */
1211 static inline bool ptlrpc_req_need_swab(struct ptlrpc_request *req)
1212 {
1213         return lustre_req_swabbed(req, MSG_PTLRPC_HEADER_OFF);
1214 }
1215
1216 /**
1217  * Returns true if request reply needs to be swabbed into local cpu byteorder
1218  */
1219 static inline bool ptlrpc_rep_need_swab(struct ptlrpc_request *req)
1220 {
1221         return lustre_rep_swabbed(req, MSG_PTLRPC_HEADER_OFF);
1222 }
1223
1224 /**
1225  * Mark request buffer at offset \a index that it was already swabbed
1226  */
1227 static inline void lustre_set_req_swabbed(struct ptlrpc_request *req,
1228                                           size_t index)
1229 {
1230         LASSERT(index < sizeof(req->rq_req_swab_mask) * 8);
1231         LASSERT((req->rq_req_swab_mask & BIT(index)) == 0);
1232         req->rq_req_swab_mask |= BIT(index);
1233 }
1234
1235 /**
1236  * Mark request reply buffer at offset \a index that it was already swabbed
1237  */
1238 static inline void lustre_set_rep_swabbed(struct ptlrpc_request *req,
1239                                           size_t index)
1240 {
1241         LASSERT(index < sizeof(req->rq_rep_swab_mask) * 8);
1242         LASSERT((req->rq_rep_swab_mask & BIT(index)) == 0);
1243         req->rq_rep_swab_mask |= BIT(index);
1244 }
1245
1246 /**
1247  * Convert numerical request phase value \a phase into text string description
1248  */
1249 static inline const char *
1250 ptlrpc_phase2str(enum rq_phase phase)
1251 {
1252         switch (phase) {
1253         case RQ_PHASE_NEW:
1254                 return "New";
1255         case RQ_PHASE_RPC:
1256                 return "Rpc";
1257         case RQ_PHASE_BULK:
1258                 return "Bulk";
1259         case RQ_PHASE_INTERPRET:
1260                 return "Interpret";
1261         case RQ_PHASE_COMPLETE:
1262                 return "Complete";
1263         case RQ_PHASE_UNREG_RPC:
1264                 return "UnregRPC";
1265         case RQ_PHASE_UNREG_BULK:
1266                 return "UnregBULK";
1267         default:
1268                 return "?Phase?";
1269         }
1270 }
1271
1272 /**
1273  * Convert numerical request phase of the request \a req into text stringi
1274  * description
1275  */
1276 static inline const char *
1277 ptlrpc_rqphase2str(struct ptlrpc_request *req)
1278 {
1279         return ptlrpc_phase2str(req->rq_phase);
1280 }
1281
1282 /**
1283  * Debugging functions and helpers to print request structure into debug log
1284  * @{
1285  */
1286 /* Spare the preprocessor, spoil the bugs. */
1287 #define FLAG(field, str) (field ? str : "")
1288
1289 /** Convert bit flags into a string */
1290 #define DEBUG_REQ_FLAGS(req)                                                   \
1291         ptlrpc_rqphase2str(req),                                               \
1292         FLAG(req->rq_intr, "I"), FLAG(req->rq_replied, "R"),                   \
1293         FLAG(req->rq_err, "E"), FLAG(req->rq_net_err, "e"),                    \
1294         FLAG(req->rq_timedout, "X") /* eXpired */, FLAG(req->rq_resend, "S"),  \
1295         FLAG(req->rq_restart, "T"), FLAG(req->rq_replay, "P"),                 \
1296         FLAG(req->rq_no_resend, "N"), FLAG(req->rq_no_reply, "n"),            \
1297         FLAG(req->rq_waiting, "W"),                                            \
1298         FLAG(req->rq_wait_ctx, "C"), FLAG(req->rq_hp, "H"),                    \
1299         FLAG(req->rq_committed, "M"),                                          \
1300         FLAG(req->rq_req_unlinked, "Q"),                                       \
1301         FLAG(req->rq_reply_unlinked, "U"),                                     \
1302         FLAG(req->rq_receiving_reply, "r")
1303
1304 #define REQ_FLAGS_FMT "%s:%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s"
1305
1306 void _debug_req(struct ptlrpc_request *req,
1307                 struct libcfs_debug_msg_data *data, const char *fmt, ...)
1308         __attribute__ ((format (printf, 3, 4)));
1309
1310 /**
1311  * Helper that decides if we need to print request accordig to current debug
1312  * level settings
1313  */
1314 #define debug_req(msgdata, mask, cdls, req, fmt, a...)                        \
1315 do {                                                                          \
1316         CFS_CHECK_STACK(msgdata, mask, cdls);                                 \
1317                                                                               \
1318         if (((mask) & D_CANTMASK) != 0 ||                                     \
1319             ((libcfs_debug & (mask)) != 0 &&                                  \
1320              (libcfs_subsystem_debug & DEBUG_SUBSYSTEM) != 0))                \
1321                 _debug_req((req), msgdata, fmt, ##a);                         \
1322 } while(0)
1323
1324 /**
1325  * This is the debug print function you need to use to print request sturucture
1326  * content into lustre debug log.
1327  * for most callers (level is a constant) this is resolved at compile time */
1328 #define DEBUG_REQ(level, req, fmt, args...)                                   \
1329 do {                                                                          \
1330         if ((level) & (D_ERROR | D_WARNING)) {                                \
1331                 static struct cfs_debug_limit_state cdls;                     \
1332                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, level, &cdls);            \
1333                 debug_req(&msgdata, level, &cdls, req, "@@@ "fmt" ", ## args);\
1334         } else {                                                              \
1335                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, level, NULL);             \
1336                 debug_req(&msgdata, level, NULL, req, "@@@ "fmt" ", ## args); \
1337         }                                                                     \
1338 } while (0)
1339 /** @} */
1340
1341 enum ptlrpc_bulk_op_type {
1342         PTLRPC_BULK_OP_ACTIVE =  0x00000001,
1343         PTLRPC_BULK_OP_PASSIVE = 0x00000002,
1344         PTLRPC_BULK_OP_PUT =     0x00000004,
1345         PTLRPC_BULK_OP_GET =     0x00000008,
1346         PTLRPC_BULK_GET_SOURCE = PTLRPC_BULK_OP_PASSIVE | PTLRPC_BULK_OP_GET,
1347         PTLRPC_BULK_PUT_SINK =   PTLRPC_BULK_OP_PASSIVE | PTLRPC_BULK_OP_PUT,
1348         PTLRPC_BULK_GET_SINK =   PTLRPC_BULK_OP_ACTIVE | PTLRPC_BULK_OP_GET,
1349         PTLRPC_BULK_PUT_SOURCE = PTLRPC_BULK_OP_ACTIVE | PTLRPC_BULK_OP_PUT,
1350 };
1351
1352 static inline bool ptlrpc_is_bulk_op_get(enum ptlrpc_bulk_op_type type)
1353 {
1354         return (type & PTLRPC_BULK_OP_GET) == PTLRPC_BULK_OP_GET;
1355 }
1356
1357 static inline bool ptlrpc_is_bulk_get_source(enum ptlrpc_bulk_op_type type)
1358 {
1359         return (type & PTLRPC_BULK_GET_SOURCE) == PTLRPC_BULK_GET_SOURCE;
1360 }
1361
1362 static inline bool ptlrpc_is_bulk_put_sink(enum ptlrpc_bulk_op_type type)
1363 {
1364         return (type & PTLRPC_BULK_PUT_SINK) == PTLRPC_BULK_PUT_SINK;
1365 }
1366
1367 static inline bool ptlrpc_is_bulk_get_sink(enum ptlrpc_bulk_op_type type)
1368 {
1369         return (type & PTLRPC_BULK_GET_SINK) == PTLRPC_BULK_GET_SINK;
1370 }
1371
1372 static inline bool ptlrpc_is_bulk_put_source(enum ptlrpc_bulk_op_type type)
1373 {
1374         return (type & PTLRPC_BULK_PUT_SOURCE) == PTLRPC_BULK_PUT_SOURCE;
1375 }
1376
1377 static inline bool ptlrpc_is_bulk_op_active(enum ptlrpc_bulk_op_type type)
1378 {
1379         return ((type & PTLRPC_BULK_OP_ACTIVE) |
1380                 (type & PTLRPC_BULK_OP_PASSIVE))
1381                         == PTLRPC_BULK_OP_ACTIVE;
1382 }
1383
1384 static inline bool ptlrpc_is_bulk_op_passive(enum ptlrpc_bulk_op_type type)
1385 {
1386         return ((type & PTLRPC_BULK_OP_ACTIVE) |
1387                 (type & PTLRPC_BULK_OP_PASSIVE))
1388                         == PTLRPC_BULK_OP_PASSIVE;
1389 }
1390
1391 struct ptlrpc_bulk_frag_ops {
1392         /**
1393          * Add a page \a page to the bulk descriptor \a desc
1394          * Data to transfer in the page starts at offset \a pageoffset and
1395          * amount of data to transfer from the page is \a len
1396          */
1397         void (*add_kiov_frag)(struct ptlrpc_bulk_desc *desc,
1398                               struct page *page, int pageoffset, int len);
1399
1400         /*
1401          * Add a \a fragment to the bulk descriptor \a desc.
1402          * Data to transfer in the fragment is pointed to by \a frag
1403          * The size of the fragment is \a len
1404          */
1405         int (*add_iov_frag)(struct ptlrpc_bulk_desc *desc, void *frag, int len);
1406
1407         /**
1408          * Uninitialize and free bulk descriptor \a desc.
1409          * Works on bulk descriptors both from server and client side.
1410          */
1411         void (*release_frags)(struct ptlrpc_bulk_desc *desc);
1412 };
1413
1414 extern const struct ptlrpc_bulk_frag_ops ptlrpc_bulk_kiov_pin_ops;
1415 extern const struct ptlrpc_bulk_frag_ops ptlrpc_bulk_kiov_nopin_ops;
1416
1417 /*
1418  * Definition of bulk descriptor.
1419  * Bulks are special "Two phase" RPCs where initial request message
1420  * is sent first and it is followed bt a transfer (o receiving) of a large
1421  * amount of data to be settled into pages referenced from the bulk descriptors.
1422  * Bulks transfers (the actual data following the small requests) are done
1423  * on separate LNet portals.
1424  * In lustre we use bulk transfers for READ and WRITE transfers from/to OSTs.
1425  *  Another user is readpage for MDT.
1426  */
1427 struct ptlrpc_bulk_desc {
1428         unsigned int    bd_refs; /* number MD's assigned including zero-sends */
1429         /** completed with failure */
1430         unsigned long bd_failure:1;
1431         /** client side */
1432         unsigned long bd_registered:1;
1433         /** For serialization with callback */
1434         spinlock_t bd_lock;
1435         /** {put,get}{source,sink}{kvec,kiov} */
1436         enum ptlrpc_bulk_op_type bd_type;
1437         /** LNet portal for this bulk */
1438         __u32 bd_portal;
1439         /** Server side - export this bulk created for */
1440         struct obd_export *bd_export;
1441         /** Client side - import this bulk was sent on */
1442         struct obd_import *bd_import;
1443         /** Back pointer to the request */
1444         struct ptlrpc_request *bd_req;
1445         const struct ptlrpc_bulk_frag_ops *bd_frag_ops;
1446         wait_queue_head_t      bd_waitq;        /* server side only WQ */
1447         int                    bd_iov_count;    /* # entries in bd_iov */
1448         int                    bd_max_iov;      /* allocated size of bd_iov */
1449         int                    bd_nob;          /* # bytes covered */
1450         int                    bd_nob_transferred; /* # bytes GOT/PUT */
1451         unsigned int            bd_nob_last;    /* # bytes in last MD */
1452
1453         __u64                  bd_last_mbits;
1454
1455         struct ptlrpc_cb_id    bd_cbid;         /* network callback info */
1456         lnet_nid_t             bd_sender;       /* stash event::sender */
1457         int                     bd_md_count;    /* # valid entries in bd_mds */
1458         int                     bd_md_max_brw;  /* max entries in bd_mds */
1459
1460         /** array of offsets for each MD */
1461         unsigned int            bd_mds_off[PTLRPC_BULK_OPS_COUNT];
1462         /** array of associated MDs */
1463         struct lnet_handle_md   bd_mds[PTLRPC_BULK_OPS_COUNT];
1464
1465         /* encrypted iov, size is either 0 or bd_iov_count. */
1466         struct bio_vec *bd_enc_vec;
1467         struct bio_vec *bd_vec;
1468 };
1469
1470 enum {
1471         SVC_INIT        = 0,
1472         SVC_STOPPED     = BIT(0),
1473         SVC_STOPPING    = BIT(1),
1474         SVC_STARTING    = BIT(2),
1475         SVC_RUNNING     = BIT(3),
1476 };
1477
1478 #define PTLRPC_THR_NAME_LEN             32
1479 /**
1480  * Definition of server service thread structure
1481  */
1482 struct ptlrpc_thread {
1483         /**
1484          * List of active threads in svcpt->scp_threads
1485          */
1486         struct list_head t_link;
1487         /**
1488          * thread-private data (preallocated vmalloc'd memory)
1489          */
1490         void *t_data;
1491         __u32 t_flags;
1492         /**
1493          * service thread index, from ptlrpc_start_threads
1494          */
1495         unsigned int t_id;
1496         /**
1497          * service thread
1498          */
1499         struct task_struct *t_task;
1500         pid_t t_pid;
1501         ktime_t t_touched;
1502         /**
1503          * put watchdog in the structure per thread b=14840
1504          */
1505         struct delayed_work t_watchdog;
1506         /**
1507          * the svc this thread belonged to b=18582
1508          */
1509         struct ptlrpc_service_part      *t_svcpt;
1510         wait_queue_head_t               t_ctl_waitq;
1511         struct lu_env                   *t_env;
1512         char                            t_name[PTLRPC_THR_NAME_LEN];
1513 };
1514
1515 static inline int thread_is_init(struct ptlrpc_thread *thread)
1516 {
1517         return thread->t_flags == 0;
1518 }
1519
1520 static inline int thread_is_stopped(struct ptlrpc_thread *thread)
1521 {
1522         return !!(thread->t_flags & SVC_STOPPED);
1523 }
1524
1525 static inline int thread_is_stopping(struct ptlrpc_thread *thread)
1526 {
1527         return !!(thread->t_flags & SVC_STOPPING);
1528 }
1529
1530 static inline int thread_is_starting(struct ptlrpc_thread *thread)
1531 {
1532         return !!(thread->t_flags & SVC_STARTING);
1533 }
1534
1535 static inline int thread_is_running(struct ptlrpc_thread *thread)
1536 {
1537         return !!(thread->t_flags & SVC_RUNNING);
1538 }
1539
1540 static inline void thread_clear_flags(struct ptlrpc_thread *thread, __u32 flags)
1541 {
1542         thread->t_flags &= ~flags;
1543 }
1544
1545 static inline void thread_set_flags(struct ptlrpc_thread *thread, __u32 flags)
1546 {
1547         thread->t_flags = flags;
1548 }
1549
1550 static inline void thread_add_flags(struct ptlrpc_thread *thread, __u32 flags)
1551 {
1552         thread->t_flags |= flags;
1553 }
1554
1555 static inline int thread_test_and_clear_flags(struct ptlrpc_thread *thread,
1556                                               __u32 flags)
1557 {
1558         if (thread->t_flags & flags) {
1559                 thread->t_flags &= ~flags;
1560                 return 1;
1561         }
1562         return 0;
1563 }
1564
1565 /**
1566  * Request buffer descriptor structure.
1567  * This is a structure that contains one posted request buffer for service.
1568  * Once data land into a buffer, event callback creates actual request and
1569  * notifies wakes one of the service threads to process new incoming request.
1570  * More than one request can fit into the buffer.
1571  */
1572 struct ptlrpc_request_buffer_desc {
1573         /** Link item for rqbds on a service */
1574         struct list_head                rqbd_list;
1575         /** History of requests for this buffer */
1576         struct list_head                rqbd_reqs;
1577         /** Back pointer to service for which this buffer is registered */
1578         struct ptlrpc_service_part      *rqbd_svcpt;
1579         /** LNet descriptor */
1580         struct lnet_handle_md           rqbd_md_h;
1581         int                             rqbd_refcount;
1582         /** The buffer itself */
1583         char                            *rqbd_buffer;
1584         struct ptlrpc_cb_id             rqbd_cbid;
1585         /**
1586          * This "embedded" request structure is only used for the
1587          * last request to fit into the buffer
1588          */
1589         struct ptlrpc_request           rqbd_req;
1590 };
1591
1592 typedef int  (*svc_handler_t)(struct ptlrpc_request *req);
1593
1594 struct ptlrpc_service_ops {
1595         /**
1596          * if non-NULL called during thread creation (ptlrpc_start_thread())
1597          * to initialize service specific per-thread state.
1598          */
1599         int             (*so_thr_init)(struct ptlrpc_thread *thr);
1600         /**
1601          * if non-NULL called during thread shutdown (ptlrpc_main()) to
1602          * destruct state created by ->srv_init().
1603          */
1604         void            (*so_thr_done)(struct ptlrpc_thread *thr);
1605         /**
1606          * Handler function for incoming requests for this service
1607          */
1608         int             (*so_req_handler)(struct ptlrpc_request *req);
1609         /**
1610          * function to determine priority of the request, it's called
1611          * on every new request
1612          */
1613         int             (*so_hpreq_handler)(struct ptlrpc_request *);
1614         /**
1615          * service-specific print fn
1616          */
1617         void            (*so_req_printer)(void *, struct ptlrpc_request *);
1618 };
1619
1620 #ifndef __cfs_cacheline_aligned
1621 /* NB: put it here for reducing patche dependence */
1622 # define __cfs_cacheline_aligned
1623 #endif
1624
1625 /**
1626  * How many high priority requests to serve before serving one normal
1627  * priority request
1628  */
1629 #define PTLRPC_SVC_HP_RATIO 10
1630
1631 /**
1632  * Definition of PortalRPC service.
1633  * The service is listening on a particular portal (like tcp port)
1634  * and perform actions for a specific server like IO service for OST
1635  * or general metadata service for MDS.
1636  */
1637 struct ptlrpc_service {
1638         /** serialize /proc operations */
1639         spinlock_t                      srv_lock;
1640         /** most often accessed fields */
1641         /** chain thru all services */
1642         struct list_head                srv_list;
1643         /** service operations table */
1644         struct ptlrpc_service_ops       srv_ops;
1645         /** only statically allocated strings here; we don't clean them */
1646         char                           *srv_name;
1647         /** only statically allocated strings here; we don't clean them */
1648         char                           *srv_thread_name;
1649         /** threads # should be created for each partition on initializing */
1650         int                             srv_nthrs_cpt_init;
1651         /** limit of threads number for each partition */
1652         int                             srv_nthrs_cpt_limit;
1653         /** Root of debugfs dir tree for this service */
1654         struct dentry                  *srv_debugfs_entry;
1655         /** Pointer to statistic data for this service */
1656         struct lprocfs_stats           *srv_stats;
1657         /** # hp per lp reqs to handle */
1658         int                             srv_hpreq_ratio;
1659         /** biggest request to receive */
1660         int                             srv_max_req_size;
1661         /** biggest reply to send */
1662         int                             srv_max_reply_size;
1663         /** size of individual buffers */
1664         int                             srv_buf_size;
1665         /** # buffers to allocate in 1 group */
1666         int                             srv_nbuf_per_group;
1667         /** Local portal on which to receive requests */
1668         __u32                           srv_req_portal;
1669         /** Portal on the client to send replies to */
1670         __u32                           srv_rep_portal;
1671         /**
1672          * Tags for lu_context associated with this thread, see struct
1673          * lu_context.
1674          */
1675         __u32                           srv_ctx_tags;
1676         /** soft watchdog timeout multiplier */
1677         int                             srv_watchdog_factor;
1678         /** under unregister_service */
1679         unsigned                        srv_is_stopping:1;
1680         /** Whether or not to restrict service threads to CPUs in this CPT */
1681         unsigned                        srv_cpt_bind:1;
1682
1683         /** max # request buffers */
1684         int                             srv_nrqbds_max;
1685         /** max # request buffers in history per partition */
1686         int                             srv_hist_nrqbds_cpt_max;
1687         /** number of CPTs this service associated with */
1688         int                             srv_ncpts;
1689         /** CPTs array this service associated with */
1690         __u32                           *srv_cpts;
1691         /** 2^srv_cptab_bits >= cfs_cpt_numbert(srv_cptable) */
1692         int                             srv_cpt_bits;
1693         /** CPT table this service is running over */
1694         struct cfs_cpt_table            *srv_cptable;
1695
1696         /* sysfs object */
1697         struct kobject                  srv_kobj;
1698         struct completion               srv_kobj_unregister;
1699         /**
1700          * partition data for ptlrpc service
1701          */
1702         struct ptlrpc_service_part      *srv_parts[0];
1703 };
1704
1705 /**
1706  * Definition of PortalRPC service partition data.
1707  * Although a service only has one instance of it right now, but we
1708  * will have multiple instances very soon (instance per CPT).
1709  *
1710  * it has four locks:
1711  * \a scp_lock
1712  *    serialize operations on rqbd and requests waiting for preprocess
1713  * \a scp_req_lock
1714  *    serialize operations active requests sent to this portal
1715  * \a scp_at_lock
1716  *    serialize adaptive timeout stuff
1717  * \a scp_rep_lock
1718  *    serialize operations on RS list (reply states)
1719  *
1720  * We don't have any use-case to take two or more locks at the same time
1721  * for now, so there is no lock order issue.
1722  */
1723 struct ptlrpc_service_part {
1724         /** back reference to owner */
1725         struct ptlrpc_service           *scp_service __cfs_cacheline_aligned;
1726         /* CPT id, reserved */
1727         int                             scp_cpt;
1728         /** always increasing number */
1729         int                             scp_thr_nextid;
1730         /** # of starting threads */
1731         int                             scp_nthrs_starting;
1732         /** # running threads */
1733         int                             scp_nthrs_running;
1734         /** service threads list */
1735         struct list_head                scp_threads;
1736
1737         /**
1738          * serialize the following fields, used for protecting
1739          * rqbd list and incoming requests waiting for preprocess,
1740          * threads starting & stopping are also protected by this lock.
1741          */
1742         spinlock_t                      scp_lock  __cfs_cacheline_aligned;
1743         /** userland serialization */
1744         struct mutex                    scp_mutex;
1745         /** total # req buffer descs allocated */
1746         int                             scp_nrqbds_total;
1747         /** # posted request buffers for receiving */
1748         int                             scp_nrqbds_posted;
1749         /** in progress of allocating rqbd */
1750         int                             scp_rqbd_allocating;
1751         /** # incoming reqs */
1752         int                             scp_nreqs_incoming;
1753         /** request buffers to be reposted */
1754         struct list_head                scp_rqbd_idle;
1755         /** req buffers receiving */
1756         struct list_head                scp_rqbd_posted;
1757         /** incoming reqs */
1758         struct list_head                scp_req_incoming;
1759         /** timeout before re-posting reqs, in jiffies */
1760         long                            scp_rqbd_timeout;
1761         /**
1762          * all threads sleep on this. This wait-queue is signalled when new
1763          * incoming request arrives and when difficult reply has to be handled.
1764          */
1765         wait_queue_head_t               scp_waitq;
1766
1767         /** request history */
1768         struct list_head                scp_hist_reqs;
1769         /** request buffer history */
1770         struct list_head                scp_hist_rqbds;
1771         /** # request buffers in history */
1772         int                             scp_hist_nrqbds;
1773         /** sequence number for request */
1774         __u64                           scp_hist_seq;
1775         /** highest seq culled from history */
1776         __u64                           scp_hist_seq_culled;
1777
1778         /**
1779          * serialize the following fields, used for processing requests
1780          * sent to this portal
1781          */
1782         spinlock_t                      scp_req_lock __cfs_cacheline_aligned;
1783         /** # reqs in either of the NRS heads below */
1784         /** # reqs being served */
1785         int                             scp_nreqs_active;
1786         /** # HPreqs being served */
1787         int                             scp_nhreqs_active;
1788         /** # hp requests handled */
1789         int                             scp_hreq_count;
1790
1791         /** NRS head for regular requests */
1792         struct ptlrpc_nrs               scp_nrs_reg;
1793         /** NRS head for HP requests; this is only valid for services that can
1794          *  handle HP requests */
1795         struct ptlrpc_nrs              *scp_nrs_hp;
1796
1797         /** AT stuff */
1798         /** @{ */
1799         /**
1800          * serialize the following fields, used for changes on
1801          * adaptive timeout
1802          */
1803         spinlock_t                      scp_at_lock __cfs_cacheline_aligned;
1804         /** estimated rpc service time */
1805         struct adaptive_timeout         scp_at_estimate;
1806         /** reqs waiting for replies */
1807         struct ptlrpc_at_array          scp_at_array;
1808         /** early reply timer */
1809         struct timer_list               scp_at_timer;
1810         /** debug */
1811         ktime_t                         scp_at_checktime;
1812         /** check early replies */
1813         unsigned                        scp_at_check;
1814         /** @} */
1815
1816         /**
1817          * serialize the following fields, used for processing
1818          * replies for this portal
1819          */
1820         spinlock_t                      scp_rep_lock __cfs_cacheline_aligned;
1821         /** all the active replies */
1822         struct list_head                scp_rep_active;
1823         /** List of free reply_states */
1824         struct list_head                scp_rep_idle;
1825         /** waitq to run, when adding stuff to srv_free_rs_list */
1826         wait_queue_head_t               scp_rep_waitq;
1827         /** # 'difficult' replies */
1828         atomic_t                        scp_nreps_difficult;
1829 };
1830
1831 #define ptlrpc_service_for_each_part(part, i, svc)                      \
1832         for (i = 0;                                                     \
1833              i < (svc)->srv_ncpts &&                                    \
1834              (svc)->srv_parts != NULL &&                                \
1835              ((part) = (svc)->srv_parts[i]) != NULL; i++)
1836
1837 /**
1838  * Declaration of ptlrpcd control structure
1839  */
1840 struct ptlrpcd_ctl {
1841         /**
1842          * Ptlrpc thread control flags (LIOD_START, LIOD_STOP, LIOD_FORCE)
1843          */
1844         unsigned long                   pc_flags;
1845         /**
1846          * Thread lock protecting structure fields.
1847          */
1848         spinlock_t                      pc_lock;
1849         /**
1850          * Start completion.
1851          */
1852         struct completion               pc_starting;
1853         /**
1854          * Stop completion.
1855          */
1856         struct completion               pc_finishing;
1857         /**
1858          * Thread requests set.
1859          */
1860         struct ptlrpc_request_set       *pc_set;
1861         /**
1862          * Thread name used in kthread_run()
1863          */
1864         char                            pc_name[16];
1865         /**
1866          * CPT the thread is bound on.
1867          */
1868         int                             pc_cpt;
1869         /**
1870          * Index of ptlrpcd thread in the array.
1871          */
1872         int                             pc_index;
1873         /**
1874          * Pointer to the array of partners' ptlrpcd_ctl structure.
1875          */
1876         struct ptlrpcd_ctl              **pc_partners;
1877         /**
1878          * Number of the ptlrpcd's partners.
1879          */
1880         int                             pc_npartners;
1881         /**
1882          * Record the partner index to be processed next.
1883          */
1884         int                             pc_cursor;
1885         /**
1886          * Error code if the thread failed to fully start.
1887          */
1888         int                             pc_error;
1889 };
1890
1891 /* Bits for pc_flags */
1892 enum ptlrpcd_ctl_flags {
1893         /**
1894          * Ptlrpc thread start flag.
1895          */
1896         LIOD_START      = BIT(0),
1897         /**
1898          * Ptlrpc thread stop flag.
1899          */
1900         LIOD_STOP       = BIT(1),
1901         /**
1902          * Ptlrpc thread force flag (only stop force so far).
1903          * This will cause aborting any inflight rpcs handled
1904          * by thread if LIOD_STOP is specified.
1905          */
1906         LIOD_FORCE      = BIT(2),
1907         /**
1908          * This is a recovery ptlrpc thread.
1909          */
1910         LIOD_RECOVERY   = BIT(3),
1911 };
1912
1913 /**
1914  * \addtogroup nrs
1915  * @{
1916  *
1917  * Service compatibility function; the policy is compatible with all services.
1918  *
1919  * \param[in] svc  The service the policy is attempting to register with.
1920  * \param[in] desc The policy descriptor
1921  *
1922  * \retval true The policy is compatible with the service
1923  *
1924  * \see ptlrpc_nrs_pol_desc::pd_compat()
1925  */
1926 static inline bool nrs_policy_compat_all(const struct ptlrpc_service *svc,
1927                                          const struct ptlrpc_nrs_pol_desc *desc)
1928 {
1929         return true;
1930 }
1931
1932 /**
1933  * Service compatibility function; the policy is compatible with only a specific
1934  * service which is identified by its human-readable name at
1935  * ptlrpc_service::srv_name.
1936  *
1937  * \param[in] svc  The service the policy is attempting to register with.
1938  * \param[in] desc The policy descriptor
1939  *
1940  * \retval false The policy is not compatible with the service
1941  * \retval true  The policy is compatible with the service
1942  *
1943  * \see ptlrpc_nrs_pol_desc::pd_compat()
1944  */
1945 static inline bool nrs_policy_compat_one(const struct ptlrpc_service *svc,
1946                                          const struct ptlrpc_nrs_pol_desc *desc)
1947 {
1948         LASSERT(desc->pd_compat_svc_name != NULL);
1949         return strcmp(svc->srv_name, desc->pd_compat_svc_name) == 0;
1950 }
1951
1952 /** @} nrs */
1953
1954 /* ptlrpc/events.c */
1955 extern int ptlrpc_uuid_to_peer(struct obd_uuid *uuid,
1956                                struct lnet_process_id *peer, lnet_nid_t *self);
1957 /**
1958  * These callbacks are invoked by LNet when something happened to
1959  * underlying buffer
1960  * @{
1961  */
1962 extern void request_out_callback(struct lnet_event *ev);
1963 extern void reply_in_callback(struct lnet_event *ev);
1964 extern void client_bulk_callback(struct lnet_event *ev);
1965 extern void request_in_callback(struct lnet_event *ev);
1966 extern void reply_out_callback(struct lnet_event *ev);
1967 #ifdef HAVE_SERVER_SUPPORT
1968 extern void server_bulk_callback(struct lnet_event *ev);
1969 #endif
1970 /** @} */
1971
1972 /* ptlrpc/connection.c */
1973 struct ptlrpc_connection *ptlrpc_connection_get(struct lnet_process_id peer,
1974                                                 lnet_nid_t self,
1975                                                 struct obd_uuid *uuid);
1976 int ptlrpc_connection_put(struct ptlrpc_connection *c);
1977 struct ptlrpc_connection *ptlrpc_connection_addref(struct ptlrpc_connection *);
1978 int ptlrpc_connection_init(void);
1979 void ptlrpc_connection_fini(void);
1980 extern lnet_pid_t ptl_get_pid(void);
1981
1982 /*
1983  * Check if the peer connection is on the local node.  We need to use GFP_NOFS
1984  * for requests from a local client to avoid recursing into the filesystem
1985  * as we might end up waiting on a page sent in the request we're serving.
1986  *
1987  * Use __GFP_HIGHMEM so that the pages can use all of the available memory
1988  * on 32-bit machines.  Use more aggressive GFP_HIGHUSER flags from non-local
1989  * clients to be able to generate more memory pressure on the OSS and allow
1990  * inactive pages to be reclaimed, since it doesn't have any other processes
1991  * or allocations that generate memory reclaim pressure.
1992  *
1993  * See b=17576 (bdf50dc9) and b=19529 (3dcf18d3) for details.
1994  */
1995 static inline bool ptlrpc_connection_is_local(struct ptlrpc_connection *conn)
1996 {
1997         if (!conn)
1998                 return false;
1999
2000         if (conn->c_peer.nid == conn->c_self)
2001                 return true;
2002
2003         RETURN(LNetIsPeerLocal(conn->c_peer.nid));
2004 }
2005
2006 /* ptlrpc/niobuf.c */
2007 /**
2008  * Actual interfacing with LNet to put/get/register/unregister stuff
2009  * @{
2010  */
2011 #ifdef HAVE_SERVER_SUPPORT
2012 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp(struct ptlrpc_request *req,
2013                                               unsigned nfrags, unsigned max_brw,
2014                                               unsigned int type,
2015                                               unsigned portal,
2016                                               const struct ptlrpc_bulk_frag_ops
2017                                                 *ops);
2018 int ptlrpc_start_bulk_transfer(struct ptlrpc_bulk_desc *desc);
2019 void ptlrpc_abort_bulk(struct ptlrpc_bulk_desc *desc);
2020
2021 static inline int ptlrpc_server_bulk_active(struct ptlrpc_bulk_desc *desc)
2022 {
2023         int rc;
2024
2025         LASSERT(desc != NULL);
2026
2027         spin_lock(&desc->bd_lock);
2028         rc = desc->bd_refs;
2029         spin_unlock(&desc->bd_lock);
2030         return rc;
2031 }
2032 #endif
2033
2034 int ptlrpc_register_bulk(struct ptlrpc_request *req);
2035 int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async);
2036
2037 static inline int ptlrpc_client_bulk_active(struct ptlrpc_request *req)
2038 {
2039         struct ptlrpc_bulk_desc *desc;
2040         int rc;
2041
2042         LASSERT(req != NULL);
2043         desc = req->rq_bulk;
2044
2045         if (!desc)
2046                 return 0;
2047
2048         if (req->rq_bulk_deadline > ktime_get_real_seconds())
2049                 return 1;
2050
2051
2052         spin_lock(&desc->bd_lock);
2053         rc = desc->bd_refs;
2054         spin_unlock(&desc->bd_lock);
2055         return rc;
2056 }
2057
2058 #define PTLRPC_REPLY_MAYBE_DIFFICULT 0x01
2059 #define PTLRPC_REPLY_EARLY           0x02
2060 int ptlrpc_send_reply(struct ptlrpc_request *req, int flags);
2061 int ptlrpc_reply(struct ptlrpc_request *req);
2062 int ptlrpc_send_error(struct ptlrpc_request *req, int difficult);
2063 int ptlrpc_error(struct ptlrpc_request *req);
2064 int ptlrpc_at_get_net_latency(struct ptlrpc_request *req);
2065 int ptl_send_rpc(struct ptlrpc_request *request, int noreply);
2066 int ptlrpc_register_rqbd(struct ptlrpc_request_buffer_desc *rqbd);
2067 /** @} */
2068
2069 /* ptlrpc/client.c */
2070 /**
2071  * Client-side portals API. Everything to send requests, receive replies,
2072  * request queues, request management, etc.
2073  * @{
2074  */
2075 void ptlrpc_request_committed(struct ptlrpc_request *req, int force);
2076
2077 void ptlrpc_init_client(int req_portal, int rep_portal, const char *name,
2078                         struct ptlrpc_client *);
2079 void ptlrpc_cleanup_client(struct obd_import *imp);
2080 struct ptlrpc_connection *ptlrpc_uuid_to_connection(struct obd_uuid *uuid,
2081                                                     lnet_nid_t nid4refnet);
2082
2083 int ptlrpc_queue_wait(struct ptlrpc_request *req);
2084 int ptlrpc_replay_req(struct ptlrpc_request *req);
2085 void ptlrpc_restart_req(struct ptlrpc_request *req);
2086 void ptlrpc_abort_inflight(struct obd_import *imp);
2087 void ptlrpc_cleanup_imp(struct obd_import *imp);
2088 void ptlrpc_abort_set(struct ptlrpc_request_set *set);
2089
2090 struct ptlrpc_request_set *ptlrpc_prep_set(void);
2091 struct ptlrpc_request_set *ptlrpc_prep_fcset(int max, set_producer_func func,
2092                                              void *arg);
2093 int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set);
2094 int ptlrpc_set_wait(const struct lu_env *env, struct ptlrpc_request_set *);
2095 void ptlrpc_set_destroy(struct ptlrpc_request_set *);
2096 void ptlrpc_set_add_req(struct ptlrpc_request_set *, struct ptlrpc_request *);
2097 #define PTLRPCD_SET ((struct ptlrpc_request_set *)1)
2098
2099 void ptlrpc_free_rq_pool(struct ptlrpc_request_pool *pool);
2100 int ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq);
2101
2102 struct ptlrpc_request_pool *
2103 ptlrpc_init_rq_pool(int, int,
2104                     int (*populate_pool)(struct ptlrpc_request_pool *, int));
2105
2106 void ptlrpc_at_set_req_timeout(struct ptlrpc_request *req);
2107 struct ptlrpc_request *ptlrpc_request_alloc(struct obd_import *imp,
2108                                             const struct req_format *format);
2109 struct ptlrpc_request *ptlrpc_request_alloc_pool(struct obd_import *imp,
2110                                             struct ptlrpc_request_pool *,
2111                                             const struct req_format *format);
2112 void ptlrpc_request_free(struct ptlrpc_request *request);
2113 int ptlrpc_request_pack(struct ptlrpc_request *request,
2114                         __u32 version, int opcode);
2115 struct ptlrpc_request *ptlrpc_request_alloc_pack(struct obd_import *imp,
2116                                                 const struct req_format *format,
2117                                                 __u32 version, int opcode);
2118 int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
2119                              __u32 version, int opcode, char **bufs,
2120                              struct ptlrpc_cli_ctx *ctx);
2121 void ptlrpc_req_finished(struct ptlrpc_request *request);
2122 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request);
2123 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req);
2124 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_imp(struct ptlrpc_request *req,
2125                                               unsigned nfrags, unsigned max_brw,
2126                                               unsigned int type,
2127                                               unsigned portal,
2128                                               const struct ptlrpc_bulk_frag_ops
2129                                                 *ops);
2130
2131 void __ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc,
2132                              struct page *page, int pageoffset, int len,
2133                              int pin);
2134
2135 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *bulk);
2136
2137 static inline void ptlrpc_release_bulk_noop(struct ptlrpc_bulk_desc *desc)
2138 {
2139 }
2140
2141 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
2142                                       struct obd_import *imp);
2143 __u64 ptlrpc_next_xid(void);
2144 __u64 ptlrpc_sample_next_xid(void);
2145 __u64 ptlrpc_req_xid(struct ptlrpc_request *request);
2146 void ptlrpc_get_mod_rpc_slot(struct ptlrpc_request *req);
2147 void ptlrpc_put_mod_rpc_slot(struct ptlrpc_request *req);
2148
2149 /* Set of routines to run a function in ptlrpcd context */
2150 void *ptlrpcd_alloc_work(struct obd_import *imp,
2151                          int (*cb)(const struct lu_env *, void *), void *data);
2152 void ptlrpcd_destroy_work(void *handler);
2153 int ptlrpcd_queue_work(void *handler);
2154
2155 /** @} */
2156 struct ptlrpc_service_buf_conf {
2157         /* nbufs is buffers # to allocate when growing the pool */
2158         unsigned int                    bc_nbufs;
2159         /* buffer size to post */
2160         unsigned int                    bc_buf_size;
2161         /* portal to listed for requests on */
2162         unsigned int                    bc_req_portal;
2163         /* portal of where to send replies to */
2164         unsigned int                    bc_rep_portal;
2165         /* maximum request size to be accepted for this service */
2166         unsigned int                    bc_req_max_size;
2167         /* maximum reply size this service can ever send */
2168         unsigned int                    bc_rep_max_size;
2169 };
2170
2171 struct ptlrpc_service_thr_conf {
2172         /* threadname should be 8 characters or less - 6 will be added on */
2173         char                            *tc_thr_name;
2174         /* threads increasing factor for each CPU */
2175         unsigned int                    tc_thr_factor;
2176         /* service threads # to start on each partition while initializing */
2177         unsigned int                    tc_nthrs_init;
2178         /*
2179          * low water of threads # upper-limit on each partition while running,
2180          * service availability may be impacted if threads number is lower
2181          * than this value. It can be ZERO if the service doesn't require
2182          * CPU affinity or there is only one partition.
2183          */
2184         unsigned int                    tc_nthrs_base;
2185         /* "soft" limit for total threads number */
2186         unsigned int                    tc_nthrs_max;
2187         /* user specified threads number, it will be validated due to
2188          * other members of this structure. */
2189         unsigned int                    tc_nthrs_user;
2190         /* bind service threads to only CPUs in their associated CPT */
2191         unsigned int                    tc_cpu_bind;
2192         /* Tags for lu_context associated with service thread */
2193         __u32                           tc_ctx_tags;
2194 };
2195
2196 struct ptlrpc_service_cpt_conf {
2197         struct cfs_cpt_table            *cc_cptable;
2198         /* string pattern to describe CPTs for a service */
2199         char                            *cc_pattern;
2200         /* whether or not to have per-CPT service partitions */
2201         bool                            cc_affinity;
2202 };
2203
2204 struct ptlrpc_service_conf {
2205         /* service name */
2206         char                            *psc_name;
2207         /* soft watchdog timeout multiplifier to print stuck service traces */
2208         unsigned int                    psc_watchdog_factor;
2209         /* buffer information */
2210         struct ptlrpc_service_buf_conf  psc_buf;
2211         /* thread information */
2212         struct ptlrpc_service_thr_conf  psc_thr;
2213         /* CPU partition information */
2214         struct ptlrpc_service_cpt_conf  psc_cpt;
2215         /* function table */
2216         struct ptlrpc_service_ops       psc_ops;
2217 };
2218
2219 /* ptlrpc/service.c */
2220 /**
2221  * Server-side services API. Register/unregister service, request state
2222  * management, service thread management
2223  *
2224  * @{
2225  */
2226 void ptlrpc_save_lock(struct ptlrpc_request *req, struct lustre_handle *lock,
2227                       int mode, bool no_ack, bool convert_lock);
2228 void ptlrpc_commit_replies(struct obd_export *exp);
2229 void ptlrpc_dispatch_difficult_reply(struct ptlrpc_reply_state *rs);
2230 void ptlrpc_schedule_difficult_reply(struct ptlrpc_reply_state *rs);
2231 int ptlrpc_hpreq_handler(struct ptlrpc_request *req);
2232 struct ptlrpc_service *ptlrpc_register_service(
2233                                 struct ptlrpc_service_conf *conf,
2234                                 struct kset *parent,
2235                                 struct dentry *debugfs_entry);
2236 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc);
2237
2238 int ptlrpc_start_threads(struct ptlrpc_service *svc);
2239 int ptlrpc_unregister_service(struct ptlrpc_service *service);
2240 int ptlrpc_service_health_check(struct ptlrpc_service *);
2241 void ptlrpc_server_drop_request(struct ptlrpc_request *req);
2242 void ptlrpc_request_change_export(struct ptlrpc_request *req,
2243                                   struct obd_export *export);
2244 void ptlrpc_update_export_timer(struct obd_export *exp,
2245                                 time64_t extra_delay);
2246
2247 int ptlrpc_hr_init(void);
2248 void ptlrpc_hr_fini(void);
2249
2250 /** @} */
2251
2252 /* ptlrpc/import.c */
2253 /**
2254  * Import API
2255  * @{
2256  */
2257 int ptlrpc_connect_import(struct obd_import *imp);
2258 int ptlrpc_connect_import_locked(struct obd_import *imp);
2259 int ptlrpc_init_import(struct obd_import *imp);
2260 int ptlrpc_disconnect_import(struct obd_import *imp, int noclose);
2261 int ptlrpc_disconnect_and_idle_import(struct obd_import *imp);
2262 int ptlrpc_import_recovery_state_machine(struct obd_import *imp);
2263 void deuuidify(char *uuid, const char *prefix, char **uuid_start,
2264                int *uuid_len);
2265 void ptlrpc_import_enter_resend(struct obd_import *imp);
2266 /* ptlrpc/pack_generic.c */
2267 int ptlrpc_reconnect_import(struct obd_import *imp);
2268 /** @} */
2269
2270 /**
2271  * ptlrpc msg buffer and swab interface
2272  *
2273  * @{
2274  */
2275 #define PTLRPC_MAX_BUFCOUNT \
2276         (sizeof(((struct ptlrpc_request *)0)->rq_req_swab_mask) * 8)
2277 #define MD_MAX_BUFLEN           (MDS_REG_MAXREQSIZE > OUT_MAXREQSIZE ? \
2278                                  MDS_REG_MAXREQSIZE : OUT_MAXREQSIZE)
2279 #define PTLRPC_MAX_BUFLEN       (OST_IO_MAXREQSIZE > MD_MAX_BUFLEN ? \
2280                                  OST_IO_MAXREQSIZE : MD_MAX_BUFLEN)
2281 bool ptlrpc_buf_need_swab(struct ptlrpc_request *req, const int inout,
2282                           __u32 index);
2283 void ptlrpc_buf_set_swabbed(struct ptlrpc_request *req, const int inout,
2284                             __u32 index);
2285 int ptlrpc_unpack_rep_msg(struct ptlrpc_request *req, int len);
2286 int ptlrpc_unpack_req_msg(struct ptlrpc_request *req, int len);
2287
2288 int lustre_msg_check_version(struct lustre_msg *msg, __u32 version);
2289 void lustre_init_msg_v2(struct lustre_msg_v2 *msg, int count, __u32 *lens,
2290                         char **bufs);
2291 int lustre_pack_request(struct ptlrpc_request *, __u32 magic, int count,
2292                         __u32 *lens, char **bufs);
2293 int lustre_pack_reply(struct ptlrpc_request *, int count, __u32 *lens,
2294                       char **bufs);
2295 int lustre_pack_reply_v2(struct ptlrpc_request *req, int count,
2296                          __u32 *lens, char **bufs, int flags);
2297 #define LPRFL_EARLY_REPLY 1
2298 int lustre_pack_reply_flags(struct ptlrpc_request *, int count, __u32 *lens,
2299                             char **bufs, int flags);
2300 int lustre_shrink_msg(struct lustre_msg *msg, int segment,
2301                       unsigned int newlen, int move_data);
2302 int lustre_grow_msg(struct lustre_msg *msg, int segment, unsigned int newlen);
2303 void lustre_free_reply_state(struct ptlrpc_reply_state *rs);
2304 int __lustre_unpack_msg(struct lustre_msg *m, int len);
2305 __u32 lustre_msg_hdr_size(__u32 magic, __u32 count);
2306 __u32 lustre_msg_size(__u32 magic, int count, __u32 *lengths);
2307 __u32 lustre_msg_size_v2(int count, __u32 *lengths);
2308 __u32 lustre_packed_msg_size(struct lustre_msg *msg);
2309 __u32 lustre_msg_early_size(void);
2310 void *lustre_msg_buf_v2(struct lustre_msg_v2 *m, __u32 n, __u32 min_size);
2311 void *lustre_msg_buf(struct lustre_msg *m, __u32 n, __u32 minlen);
2312 __u32 lustre_msg_buflen(struct lustre_msg *m, __u32 n);
2313 void lustre_msg_set_buflen(struct lustre_msg *m, __u32 n, __u32 len);
2314 __u32 lustre_msg_bufcount(struct lustre_msg *m);
2315 char *lustre_msg_string(struct lustre_msg *m, __u32 n, __u32 max_len);
2316 __u32 lustre_msghdr_get_flags(struct lustre_msg *msg);
2317 void lustre_msghdr_set_flags(struct lustre_msg *msg, __u32 flags);
2318 __u32 lustre_msg_get_flags(struct lustre_msg *msg);
2319 void lustre_msg_add_flags(struct lustre_msg *msg, __u32 flags);
2320 void lustre_msg_set_flags(struct lustre_msg *msg, __u32 flags);
2321 void lustre_msg_clear_flags(struct lustre_msg *msg, __u32 flags);
2322 __u32 lustre_msg_get_op_flags(struct lustre_msg *msg);
2323 void lustre_msg_add_op_flags(struct lustre_msg *msg, __u32 flags);
2324 struct lustre_handle *lustre_msg_get_handle(struct lustre_msg *msg);
2325 __u32 lustre_msg_get_type(struct lustre_msg *msg);
2326 enum lustre_msg_version lustre_msg_get_version(struct lustre_msg *msg);
2327 void lustre_msg_add_version(struct lustre_msg *msg, __u32 version);
2328 __u32 lustre_msg_get_opc(struct lustre_msg *msg);
2329 __u64 lustre_msg_get_last_xid(struct lustre_msg *msg);
2330 __u16 lustre_msg_get_tag(struct lustre_msg *msg);
2331 __u64 lustre_msg_get_last_committed(struct lustre_msg *msg);
2332 __u64 *lustre_msg_get_versions(struct lustre_msg *msg);
2333 __u64 lustre_msg_get_transno(struct lustre_msg *msg);
2334 __u64 lustre_msg_get_slv(struct lustre_msg *msg);
2335 __u32 lustre_msg_get_limit(struct lustre_msg *msg);
2336 void lustre_msg_set_slv(struct lustre_msg *msg, __u64 slv);
2337 void lustre_msg_set_limit(struct lustre_msg *msg, __u64 limit);
2338 int lustre_msg_get_status(struct lustre_msg *msg);
2339 __u32 lustre_msg_get_conn_cnt(struct lustre_msg *msg);
2340 __u32 lustre_msg_get_magic(struct lustre_msg *msg);
2341 timeout_t lustre_msg_get_timeout(struct lustre_msg *msg);
2342 timeout_t lustre_msg_get_service_timeout(struct lustre_msg *msg);
2343 char *lustre_msg_get_jobid(struct lustre_msg *msg);
2344 __u32 lustre_msg_get_cksum(struct lustre_msg *msg);
2345 __u64 lustre_msg_get_mbits(struct lustre_msg *msg);
2346 __u32 lustre_msg_calc_cksum(struct lustre_msg *msg);
2347 void lustre_msg_set_handle(struct lustre_msg *msg,struct lustre_handle *handle);
2348 void lustre_msg_set_type(struct lustre_msg *msg, __u32 type);
2349 void lustre_msg_set_opc(struct lustre_msg *msg, __u32 opc);
2350 void lustre_msg_set_last_xid(struct lustre_msg *msg, __u64 last_xid);
2351 void lustre_msg_set_tag(struct lustre_msg *msg, __u16 tag);
2352 void lustre_msg_set_last_committed(struct lustre_msg *msg,__u64 last_committed);
2353 void lustre_msg_set_versions(struct lustre_msg *msg, __u64 *versions);
2354 void lustre_msg_set_transno(struct lustre_msg *msg, __u64 transno);
2355 void lustre_msg_set_status(struct lustre_msg *msg, __u32 status);
2356 void lustre_msg_set_conn_cnt(struct lustre_msg *msg, __u32 conn_cnt);
2357 void ptlrpc_req_set_repsize(struct ptlrpc_request *req, int count, __u32 *sizes);
2358 void ptlrpc_request_set_replen(struct ptlrpc_request *req);
2359 void lustre_msg_set_timeout(struct lustre_msg *msg, timeout_t timeout);
2360 void lustre_msg_set_service_timeout(struct lustre_msg *msg,
2361                                     timeout_t service_timeout);
2362 void lustre_msg_set_jobid(struct lustre_msg *msg, char *jobid);
2363 void lustre_msg_set_cksum(struct lustre_msg *msg, __u32 cksum);
2364 void lustre_msg_set_mbits(struct lustre_msg *msg, __u64 mbits);
2365
2366 static inline void
2367 lustre_shrink_reply(struct ptlrpc_request *req, int segment,
2368                     unsigned int newlen, int move_data)
2369 {
2370         LASSERT(req->rq_reply_state);
2371         LASSERT(req->rq_repmsg);
2372         req->rq_replen = lustre_shrink_msg(req->rq_repmsg, segment,
2373                                            newlen, move_data);
2374 }
2375
2376 #ifdef LUSTRE_TRANSLATE_ERRNOS
2377
2378 static inline int ptlrpc_status_hton(int h)
2379 {
2380         /*
2381          * Positive errnos must be network errnos, such as LUSTRE_EDEADLK,
2382          * ELDLM_LOCK_ABORTED, etc.
2383          */
2384         if (h < 0)
2385                 return -lustre_errno_hton(-h);
2386         else
2387                 return h;
2388 }
2389
2390 static inline int ptlrpc_status_ntoh(int n)
2391 {
2392         /*
2393          * See the comment in ptlrpc_status_hton().
2394          */
2395         if (n < 0)
2396                 return -lustre_errno_ntoh(-n);
2397         else
2398                 return n;
2399 }
2400
2401 #else
2402
2403 #define ptlrpc_status_hton(h) (h)
2404 #define ptlrpc_status_ntoh(n) (n)
2405
2406 #endif
2407 /** @} */
2408
2409 /** Change request phase of \a req to \a new_phase */
2410 static inline void
2411 ptlrpc_rqphase_move(struct ptlrpc_request *req, enum rq_phase new_phase)
2412 {
2413         if (req->rq_phase == new_phase)
2414                 return;
2415
2416         if (new_phase == RQ_PHASE_UNREG_RPC ||
2417             new_phase == RQ_PHASE_UNREG_BULK) {
2418                 /* No embedded unregistering phases */
2419                 if (req->rq_phase == RQ_PHASE_UNREG_RPC ||
2420                     req->rq_phase == RQ_PHASE_UNREG_BULK)
2421                         return;
2422
2423                 req->rq_next_phase = req->rq_phase;
2424                 if (req->rq_import)
2425                         atomic_inc(&req->rq_import->imp_unregistering);
2426         }
2427
2428         if (req->rq_phase == RQ_PHASE_UNREG_RPC ||
2429             req->rq_phase == RQ_PHASE_UNREG_BULK) {
2430                 if (req->rq_import)
2431                         atomic_dec(&req->rq_import->imp_unregistering);
2432         }
2433
2434         DEBUG_REQ(D_INFO, req, "move request phase from %s to %s",
2435                   ptlrpc_rqphase2str(req), ptlrpc_phase2str(new_phase));
2436
2437         req->rq_phase = new_phase;
2438 }
2439
2440 /**
2441  * Returns true if request \a req got early reply and hard deadline is not met
2442  */
2443 static inline int
2444 ptlrpc_client_early(struct ptlrpc_request *req)
2445 {
2446         return req->rq_early;
2447 }
2448
2449 /**
2450  * Returns true if we got real reply from server for this request
2451  */
2452 static inline int
2453 ptlrpc_client_replied(struct ptlrpc_request *req)
2454 {
2455         if (req->rq_reply_deadline > ktime_get_real_seconds())
2456                 return 0;
2457         return req->rq_replied;
2458 }
2459
2460 /** Returns true if request \a req is in process of receiving server reply */
2461 static inline int
2462 ptlrpc_client_recv(struct ptlrpc_request *req)
2463 {
2464         if (req->rq_reply_deadline > ktime_get_real_seconds())
2465                 return 1;
2466         return req->rq_receiving_reply;
2467 }
2468
2469 static inline int
2470 ptlrpc_client_recv_or_unlink(struct ptlrpc_request *req)
2471 {
2472         int rc;
2473
2474         spin_lock(&req->rq_lock);
2475         if (req->rq_reply_deadline > ktime_get_real_seconds()) {
2476                 spin_unlock(&req->rq_lock);
2477                 return 1;
2478         }
2479         if (req->rq_req_deadline > ktime_get_real_seconds()) {
2480                 spin_unlock(&req->rq_lock);
2481                 return 1;
2482         }
2483
2484         rc = !req->rq_req_unlinked || !req->rq_reply_unlinked ||
2485              req->rq_receiving_reply;
2486         spin_unlock(&req->rq_lock);
2487         return rc;
2488 }
2489
2490 static inline void
2491 ptlrpc_client_wake_req(struct ptlrpc_request *req)
2492 {
2493         smp_mb();
2494         if (req->rq_set == NULL)
2495                 wake_up(&req->rq_reply_waitq);
2496         else
2497                 wake_up(&req->rq_set->set_waitq);
2498 }
2499
2500 static inline void
2501 ptlrpc_rs_addref(struct ptlrpc_reply_state *rs)
2502 {
2503         LASSERT(atomic_read(&rs->rs_refcount) > 0);
2504         atomic_inc(&rs->rs_refcount);
2505 }
2506
2507 static inline void
2508 ptlrpc_rs_decref(struct ptlrpc_reply_state *rs)
2509 {
2510         LASSERT(atomic_read(&rs->rs_refcount) > 0);
2511         if (atomic_dec_and_test(&rs->rs_refcount))
2512                 lustre_free_reply_state(rs);
2513 }
2514
2515 /* Should only be called once per req */
2516 static inline void ptlrpc_req_drop_rs(struct ptlrpc_request *req)
2517 {
2518         if (req->rq_reply_state == NULL)
2519                 return; /* shouldn't occur */
2520         ptlrpc_rs_decref(req->rq_reply_state);
2521         req->rq_reply_state = NULL;
2522         req->rq_repmsg = NULL;
2523 }
2524
2525 static inline __u32 lustre_request_magic(struct ptlrpc_request *req)
2526 {
2527         return lustre_msg_get_magic(req->rq_reqmsg);
2528 }
2529
2530 static inline int ptlrpc_req_get_repsize(struct ptlrpc_request *req)
2531 {
2532         switch (req->rq_reqmsg->lm_magic) {
2533         case LUSTRE_MSG_MAGIC_V2:
2534                 return req->rq_reqmsg->lm_repsize;
2535         default:
2536                 LASSERTF(0, "incorrect message magic: %08x\n",
2537                          req->rq_reqmsg->lm_magic);
2538                 return -EFAULT;
2539         }
2540 }
2541
2542 static inline int ptlrpc_send_limit_expired(struct ptlrpc_request *req)
2543 {
2544         if (req->rq_delay_limit != 0 &&
2545             req->rq_queued_time + req->rq_delay_limit < ktime_get_seconds())
2546                 return 1;
2547         return 0;
2548 }
2549
2550 static inline int ptlrpc_no_resend(struct ptlrpc_request *req)
2551 {
2552         if (!req->rq_no_resend && ptlrpc_send_limit_expired(req)) {
2553                 spin_lock(&req->rq_lock);
2554                 req->rq_no_resend = 1;
2555                 spin_unlock(&req->rq_lock);
2556         }
2557         return req->rq_no_resend;
2558 }
2559
2560 static inline int
2561 ptlrpc_server_get_timeout(struct ptlrpc_service_part *svcpt)
2562 {
2563         int at = AT_OFF ? 0 : at_get(&svcpt->scp_at_estimate);
2564
2565         return svcpt->scp_service->srv_watchdog_factor *
2566                max_t(int, at, obd_timeout);
2567 }
2568
2569 /**
2570  * Calculate the amount of time for lock prolongation.
2571  *
2572  * This is helper function to get the timeout extra time.
2573  *
2574  * @req         current request
2575  *
2576  * Return:      amount of time to extend the timeout with
2577  */
2578 static inline timeout_t prolong_timeout(struct ptlrpc_request *req)
2579 {
2580         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
2581         timeout_t req_timeout = 0;
2582
2583         if (AT_OFF)
2584                 return obd_timeout / 2;
2585
2586         if (req->rq_deadline > req->rq_arrival_time.tv_sec)
2587                 req_timeout = req->rq_deadline - req->rq_arrival_time.tv_sec;
2588
2589         return max(req_timeout,
2590                    at_est2timeout(at_get(&svcpt->scp_at_estimate)));
2591 }
2592
2593 static inline struct ptlrpc_service *
2594 ptlrpc_req2svc(struct ptlrpc_request *req)
2595 {
2596         LASSERT(req->rq_rqbd != NULL);
2597         return req->rq_rqbd->rqbd_svcpt->scp_service;
2598 }
2599
2600 /* ldlm/ldlm_lib.c */
2601 /**
2602  * Target client logic
2603  * @{
2604  */
2605 int client_obd_setup(struct obd_device *obd, struct lustre_cfg *lcfg);
2606 int client_obd_cleanup(struct obd_device *obd);
2607 int client_connect_import(const struct lu_env *env,
2608                           struct obd_export **exp, struct obd_device *obd,
2609                           struct obd_uuid *cluuid, struct obd_connect_data *,
2610                           void *localdata);
2611 int client_disconnect_export(struct obd_export *exp);
2612 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
2613                            int priority);
2614 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid);
2615 int client_import_find_conn(struct obd_import *imp, lnet_nid_t peer,
2616                             struct obd_uuid *uuid);
2617 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid);
2618 void client_destroy_import(struct obd_import *imp);
2619 /** @} */
2620
2621 #ifdef HAVE_SERVER_SUPPORT
2622 int server_disconnect_export(struct obd_export *exp);
2623 #endif
2624
2625 /* ptlrpc/pinger.c */
2626 /**
2627  * Pinger API (client side only)
2628  * @{
2629  */
2630 enum timeout_event {
2631         TIMEOUT_GRANT = 1
2632 };
2633 struct timeout_item;
2634 typedef int (*timeout_cb_t)(struct timeout_item *, void *);
2635 int ptlrpc_pinger_add_import(struct obd_import *imp);
2636 int ptlrpc_pinger_del_import(struct obd_import *imp);
2637 int ptlrpc_add_timeout_client(time64_t time, enum timeout_event event,
2638                               timeout_cb_t cb, void *data,
2639                               struct list_head *obd_list);
2640 int ptlrpc_del_timeout_client(struct list_head *obd_list,
2641                               enum timeout_event event);
2642 struct ptlrpc_request * ptlrpc_prep_ping(struct obd_import *imp);
2643 int ptlrpc_obd_ping(struct obd_device *obd);
2644 void ping_evictor_start(void);
2645 void ping_evictor_stop(void);
2646 void ptlrpc_pinger_ir_up(void);
2647 void ptlrpc_pinger_ir_down(void);
2648 /** @} */
2649 int ptlrpc_pinger_suppress_pings(void);
2650
2651 /* ptlrpc/ptlrpcd.c */
2652 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force);
2653 void ptlrpcd_free(struct ptlrpcd_ctl *pc);
2654 void ptlrpcd_wake(struct ptlrpc_request *req);
2655 void ptlrpcd_add_req(struct ptlrpc_request *req);
2656 void ptlrpcd_add_rqset(struct ptlrpc_request_set *set);
2657 int ptlrpcd_addref(void);
2658 void ptlrpcd_decref(void);
2659
2660 /* ptlrpc/lproc_ptlrpc.c */
2661 /**
2662  * procfs output related functions
2663  * @{
2664  */
2665 const char* ll_opcode2str(__u32 opcode);
2666 const int ll_str2opcode(const char *ops);
2667 #ifdef CONFIG_PROC_FS
2668 void ptlrpc_lprocfs_register_obd(struct obd_device *obd);
2669 void ptlrpc_lprocfs_unregister_obd(struct obd_device *obd);
2670 void ptlrpc_lprocfs_brw(struct ptlrpc_request *req, int bytes);
2671 #else
2672 static inline void ptlrpc_lprocfs_register_obd(struct obd_device *obd) {}
2673 static inline void ptlrpc_lprocfs_unregister_obd(struct obd_device *obd) {}
2674 static inline void ptlrpc_lprocfs_brw(struct ptlrpc_request *req, int bytes) {}
2675 #endif
2676 /** @} */
2677
2678 /* ptlrpc/llog_server.c */
2679 int llog_origin_handle_open(struct ptlrpc_request *req);
2680 int llog_origin_handle_prev_block(struct ptlrpc_request *req);
2681 int llog_origin_handle_next_block(struct ptlrpc_request *req);
2682 int llog_origin_handle_read_header(struct ptlrpc_request *req);
2683
2684 /* ptlrpc/llog_client.c */
2685 extern struct llog_operations llog_client_ops;
2686 /** @} net */
2687
2688 #endif
2689 /** @} PtlRPC */