Whamcloud - gitweb
LU-9679 ptlrpc: use OBD_ALLOC_PTR_ARRAY() and FREE
[fs/lustre-release.git] / lustre / ptlrpc / service.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 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
33 #define DEBUG_SUBSYSTEM S_RPC
34
35 #include <linux/kthread.h>
36 #include <linux/ratelimit.h>
37
38 #include <obd_support.h>
39 #include <obd_class.h>
40 #include <lustre_net.h>
41 #include <lu_object.h>
42 #include <uapi/linux/lnet/lnet-types.h>
43 #include "ptlrpc_internal.h"
44 #include <linux/delay.h>
45
46 /* The following are visible and mutable through /sys/module/ptlrpc */
47 int test_req_buffer_pressure = 0;
48 module_param(test_req_buffer_pressure, int, 0444);
49 MODULE_PARM_DESC(test_req_buffer_pressure, "set non-zero to put pressure on request buffer pools");
50 module_param(at_min, int, 0644);
51 MODULE_PARM_DESC(at_min, "Adaptive timeout minimum (sec)");
52 module_param(at_max, int, 0644);
53 MODULE_PARM_DESC(at_max, "Adaptive timeout maximum (sec)");
54 module_param(at_history, int, 0644);
55 MODULE_PARM_DESC(at_history,
56                  "Adaptive timeouts remember the slowest event that took place within this period (sec)");
57 module_param(at_early_margin, int, 0644);
58 MODULE_PARM_DESC(at_early_margin, "How soon before an RPC deadline to send an early reply");
59 module_param(at_extra, int, 0644);
60 MODULE_PARM_DESC(at_extra, "How much extra time to give with each early reply");
61
62 /* forward ref */
63 static int ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt);
64 static void ptlrpc_server_hpreq_fini(struct ptlrpc_request *req);
65 static void ptlrpc_at_remove_timed(struct ptlrpc_request *req);
66
67 /** Holds a list of all PTLRPC services */
68 LIST_HEAD(ptlrpc_all_services);
69 /** Used to protect the \e ptlrpc_all_services list */
70 struct mutex ptlrpc_all_services_mutex;
71
72 static struct ptlrpc_request_buffer_desc *
73 ptlrpc_alloc_rqbd(struct ptlrpc_service_part *svcpt)
74 {
75         struct ptlrpc_service             *svc = svcpt->scp_service;
76         struct ptlrpc_request_buffer_desc *rqbd;
77
78         OBD_CPT_ALLOC_PTR(rqbd, svc->srv_cptable, svcpt->scp_cpt);
79         if (rqbd == NULL)
80                 return NULL;
81
82         rqbd->rqbd_svcpt = svcpt;
83         rqbd->rqbd_refcount = 0;
84         rqbd->rqbd_cbid.cbid_fn = request_in_callback;
85         rqbd->rqbd_cbid.cbid_arg = rqbd;
86         INIT_LIST_HEAD(&rqbd->rqbd_reqs);
87         OBD_CPT_ALLOC_LARGE(rqbd->rqbd_buffer, svc->srv_cptable,
88                             svcpt->scp_cpt, svc->srv_buf_size);
89         if (rqbd->rqbd_buffer == NULL) {
90                 OBD_FREE_PTR(rqbd);
91                 return NULL;
92         }
93
94         spin_lock(&svcpt->scp_lock);
95         list_add(&rqbd->rqbd_list, &svcpt->scp_rqbd_idle);
96         svcpt->scp_nrqbds_total++;
97         spin_unlock(&svcpt->scp_lock);
98
99         return rqbd;
100 }
101
102 static void ptlrpc_free_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
103 {
104         struct ptlrpc_service_part *svcpt = rqbd->rqbd_svcpt;
105
106         LASSERT(rqbd->rqbd_refcount == 0);
107         LASSERT(list_empty(&rqbd->rqbd_reqs));
108
109         spin_lock(&svcpt->scp_lock);
110         list_del(&rqbd->rqbd_list);
111         svcpt->scp_nrqbds_total--;
112         spin_unlock(&svcpt->scp_lock);
113
114         OBD_FREE_LARGE(rqbd->rqbd_buffer, svcpt->scp_service->srv_buf_size);
115         OBD_FREE_PTR(rqbd);
116 }
117
118 static int ptlrpc_grow_req_bufs(struct ptlrpc_service_part *svcpt, int post)
119 {
120         struct ptlrpc_service *svc = svcpt->scp_service;
121         struct ptlrpc_request_buffer_desc *rqbd;
122         int rc = 0;
123         int i;
124
125         if (svcpt->scp_rqbd_allocating)
126                 goto try_post;
127
128         spin_lock(&svcpt->scp_lock);
129         /* check again with lock */
130         if (svcpt->scp_rqbd_allocating) {
131                 /* NB: we might allow more than one thread in the future */
132                 LASSERT(svcpt->scp_rqbd_allocating == 1);
133                 spin_unlock(&svcpt->scp_lock);
134                 goto try_post;
135         }
136
137         svcpt->scp_rqbd_allocating++;
138         spin_unlock(&svcpt->scp_lock);
139
140
141         for (i = 0; i < svc->srv_nbuf_per_group; i++) {
142                 /*
143                  * NB: another thread might have recycled enough rqbds, we
144                  * need to make sure it wouldn't over-allocate, see LU-1212.
145                  */
146                 if (svcpt->scp_nrqbds_posted >= svc->srv_nbuf_per_group ||
147                     (svc->srv_nrqbds_max != 0 &&
148                      svcpt->scp_nrqbds_total > svc->srv_nrqbds_max))
149                         break;
150
151                 rqbd = ptlrpc_alloc_rqbd(svcpt);
152
153                 if (rqbd == NULL) {
154                         CERROR("%s: Can't allocate request buffer\n",
155                                svc->srv_name);
156                         rc = -ENOMEM;
157                         break;
158                 }
159         }
160
161         spin_lock(&svcpt->scp_lock);
162
163         LASSERT(svcpt->scp_rqbd_allocating == 1);
164         svcpt->scp_rqbd_allocating--;
165
166         spin_unlock(&svcpt->scp_lock);
167
168         CDEBUG(D_RPCTRACE,
169                "%s: allocate %d new %d-byte reqbufs (%d/%d left), rc = %d\n",
170                svc->srv_name, i, svc->srv_buf_size, svcpt->scp_nrqbds_posted,
171                svcpt->scp_nrqbds_total, rc);
172
173  try_post:
174         if (post && rc == 0)
175                 rc = ptlrpc_server_post_idle_rqbds(svcpt);
176
177         return rc;
178 }
179
180 /**
181  * Part of Rep-Ack logic.
182  * Puts a lock and its mode into reply state assotiated to request reply.
183  */
184 void ptlrpc_save_lock(struct ptlrpc_request *req, struct lustre_handle *lock,
185                       int mode, bool no_ack, bool convert_lock)
186 {
187         struct ptlrpc_reply_state *rs = req->rq_reply_state;
188         int idx;
189
190         LASSERT(rs != NULL);
191         LASSERT(rs->rs_nlocks < RS_MAX_LOCKS);
192
193         idx = rs->rs_nlocks++;
194         rs->rs_locks[idx] = *lock;
195         rs->rs_modes[idx] = mode;
196         rs->rs_difficult = 1;
197         rs->rs_no_ack = no_ack;
198         rs->rs_convert_lock = convert_lock;
199 }
200 EXPORT_SYMBOL(ptlrpc_save_lock);
201
202
203 struct ptlrpc_hr_partition;
204
205 struct ptlrpc_hr_thread {
206         int                             hrt_id;         /* thread ID */
207         spinlock_t                      hrt_lock;
208         wait_queue_head_t               hrt_waitq;
209         struct list_head                hrt_queue;
210         struct ptlrpc_hr_partition      *hrt_partition;
211 };
212
213 struct ptlrpc_hr_partition {
214         /* # of started threads */
215         atomic_t                        hrp_nstarted;
216         /* # of stopped threads */
217         atomic_t                        hrp_nstopped;
218         /* cpu partition id */
219         int                             hrp_cpt;
220         /* round-robin rotor for choosing thread */
221         int                             hrp_rotor;
222         /* total number of threads on this partition */
223         int                             hrp_nthrs;
224         /* threads table */
225         struct ptlrpc_hr_thread         *hrp_thrs;
226 };
227
228 #define HRT_RUNNING 0
229 #define HRT_STOPPING 1
230
231 struct ptlrpc_hr_service {
232         /* CPU partition table, it's just cfs_cpt_tab for now */
233         struct cfs_cpt_table            *hr_cpt_table;
234         /** controller sleep waitq */
235         wait_queue_head_t               hr_waitq;
236         unsigned int                    hr_stopping;
237         /** roundrobin rotor for non-affinity service */
238         unsigned int                    hr_rotor;
239         /* partition data */
240         struct ptlrpc_hr_partition      **hr_partitions;
241 };
242
243 struct rs_batch {
244         struct list_head                        rsb_replies;
245         unsigned int                    rsb_n_replies;
246         struct ptlrpc_service_part      *rsb_svcpt;
247 };
248
249 /** reply handling service. */
250 static struct ptlrpc_hr_service         ptlrpc_hr;
251
252 /**
253  * maximum mumber of replies scheduled in one batch
254  */
255 #define MAX_SCHEDULED 256
256
257 /**
258  * Initialize a reply batch.
259  *
260  * \param b batch
261  */
262 static void rs_batch_init(struct rs_batch *b)
263 {
264         memset(b, 0, sizeof(*b));
265         INIT_LIST_HEAD(&b->rsb_replies);
266 }
267
268 /**
269  * Choose an hr thread to dispatch requests to.
270  */
271 static
272 struct ptlrpc_hr_thread *ptlrpc_hr_select(struct ptlrpc_service_part *svcpt)
273 {
274         struct ptlrpc_hr_partition      *hrp;
275         unsigned int                    rotor;
276
277         if (svcpt->scp_cpt >= 0 &&
278             svcpt->scp_service->srv_cptable == ptlrpc_hr.hr_cpt_table) {
279                 /* directly match partition */
280                 hrp = ptlrpc_hr.hr_partitions[svcpt->scp_cpt];
281
282         } else {
283                 rotor = ptlrpc_hr.hr_rotor++;
284                 rotor %= cfs_cpt_number(ptlrpc_hr.hr_cpt_table);
285
286                 hrp = ptlrpc_hr.hr_partitions[rotor];
287         }
288
289         rotor = hrp->hrp_rotor++;
290         return &hrp->hrp_thrs[rotor % hrp->hrp_nthrs];
291 }
292
293 /**
294  * Dispatch all replies accumulated in the batch to one from
295  * dedicated reply handling threads.
296  *
297  * \param b batch
298  */
299 static void rs_batch_dispatch(struct rs_batch *b)
300 {
301         if (b->rsb_n_replies != 0) {
302                 struct ptlrpc_hr_thread *hrt;
303
304                 hrt = ptlrpc_hr_select(b->rsb_svcpt);
305
306                 spin_lock(&hrt->hrt_lock);
307                 list_splice_init(&b->rsb_replies, &hrt->hrt_queue);
308                 spin_unlock(&hrt->hrt_lock);
309
310                 wake_up(&hrt->hrt_waitq);
311                 b->rsb_n_replies = 0;
312         }
313 }
314
315 /**
316  * Add a reply to a batch.
317  * Add one reply object to a batch, schedule batched replies if overload.
318  *
319  * \param b batch
320  * \param rs reply
321  */
322 static void rs_batch_add(struct rs_batch *b, struct ptlrpc_reply_state *rs)
323 {
324         struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
325
326         if (svcpt != b->rsb_svcpt || b->rsb_n_replies >= MAX_SCHEDULED) {
327                 if (b->rsb_svcpt != NULL) {
328                         rs_batch_dispatch(b);
329                         spin_unlock(&b->rsb_svcpt->scp_rep_lock);
330                 }
331                 spin_lock(&svcpt->scp_rep_lock);
332                 b->rsb_svcpt = svcpt;
333         }
334         spin_lock(&rs->rs_lock);
335         rs->rs_scheduled_ever = 1;
336         if (rs->rs_scheduled == 0) {
337                 list_move(&rs->rs_list, &b->rsb_replies);
338                 rs->rs_scheduled = 1;
339                 b->rsb_n_replies++;
340         }
341         rs->rs_committed = 1;
342         spin_unlock(&rs->rs_lock);
343 }
344
345 /**
346  * Reply batch finalization.
347  * Dispatch remaining replies from the batch
348  * and release remaining spinlock.
349  *
350  * \param b batch
351  */
352 static void rs_batch_fini(struct rs_batch *b)
353 {
354         if (b->rsb_svcpt != NULL) {
355                 rs_batch_dispatch(b);
356                 spin_unlock(&b->rsb_svcpt->scp_rep_lock);
357         }
358 }
359
360 #define DECLARE_RS_BATCH(b)     struct rs_batch b
361
362
363 /**
364  * Put reply state into a queue for processing because we received
365  * ACK from the client
366  */
367 void ptlrpc_dispatch_difficult_reply(struct ptlrpc_reply_state *rs)
368 {
369         struct ptlrpc_hr_thread *hrt;
370
371         ENTRY;
372
373         LASSERT(list_empty(&rs->rs_list));
374
375         hrt = ptlrpc_hr_select(rs->rs_svcpt);
376
377         spin_lock(&hrt->hrt_lock);
378         list_add_tail(&rs->rs_list, &hrt->hrt_queue);
379         spin_unlock(&hrt->hrt_lock);
380
381         wake_up(&hrt->hrt_waitq);
382         EXIT;
383 }
384
385 void ptlrpc_schedule_difficult_reply(struct ptlrpc_reply_state *rs)
386 {
387         ENTRY;
388
389         assert_spin_locked(&rs->rs_svcpt->scp_rep_lock);
390         assert_spin_locked(&rs->rs_lock);
391         LASSERT(rs->rs_difficult);
392         rs->rs_scheduled_ever = 1;  /* flag any notification attempt */
393
394         if (rs->rs_scheduled) {     /* being set up or already notified */
395                 EXIT;
396                 return;
397         }
398
399         rs->rs_scheduled = 1;
400         list_del_init(&rs->rs_list);
401         ptlrpc_dispatch_difficult_reply(rs);
402         EXIT;
403 }
404 EXPORT_SYMBOL(ptlrpc_schedule_difficult_reply);
405
406 void ptlrpc_commit_replies(struct obd_export *exp)
407 {
408         struct ptlrpc_reply_state *rs, *nxt;
409         DECLARE_RS_BATCH(batch);
410
411         ENTRY;
412
413         rs_batch_init(&batch);
414         /*
415          * Find any replies that have been committed and get their service
416          * to attend to complete them.
417          */
418
419         /* CAVEAT EMPTOR: spinlock ordering!!! */
420         spin_lock(&exp->exp_uncommitted_replies_lock);
421         list_for_each_entry_safe(rs, nxt, &exp->exp_uncommitted_replies,
422                                  rs_obd_list) {
423                 LASSERT(rs->rs_difficult);
424                 /* VBR: per-export last_committed */
425                 LASSERT(rs->rs_export);
426                 if (rs->rs_transno <= exp->exp_last_committed) {
427                         list_del_init(&rs->rs_obd_list);
428                         rs_batch_add(&batch, rs);
429                 }
430         }
431         spin_unlock(&exp->exp_uncommitted_replies_lock);
432         rs_batch_fini(&batch);
433         EXIT;
434 }
435
436 static int ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt)
437 {
438         struct ptlrpc_request_buffer_desc *rqbd;
439         int rc;
440         int posted = 0;
441
442         for (;;) {
443                 spin_lock(&svcpt->scp_lock);
444
445                 if (list_empty(&svcpt->scp_rqbd_idle)) {
446                         spin_unlock(&svcpt->scp_lock);
447                         return posted;
448                 }
449
450                 rqbd = list_entry(svcpt->scp_rqbd_idle.next,
451                                       struct ptlrpc_request_buffer_desc,
452                                       rqbd_list);
453
454                 /* assume we will post successfully */
455                 svcpt->scp_nrqbds_posted++;
456                 list_move(&rqbd->rqbd_list, &svcpt->scp_rqbd_posted);
457
458                 spin_unlock(&svcpt->scp_lock);
459
460                 rc = ptlrpc_register_rqbd(rqbd);
461                 if (rc != 0)
462                         break;
463
464                 posted = 1;
465         }
466
467         spin_lock(&svcpt->scp_lock);
468
469         svcpt->scp_nrqbds_posted--;
470         list_move_tail(&rqbd->rqbd_list, &svcpt->scp_rqbd_idle);
471
472         /*
473          * Don't complain if no request buffers are posted right now; LNET
474          * won't drop requests because we set the portal lazy!
475          */
476
477         spin_unlock(&svcpt->scp_lock);
478
479         return -1;
480 }
481
482 static void ptlrpc_at_timer(cfs_timer_cb_arg_t data)
483 {
484         struct ptlrpc_service_part *svcpt;
485
486         svcpt = cfs_from_timer(svcpt, data, scp_at_timer);
487
488         svcpt->scp_at_check = 1;
489         svcpt->scp_at_checktime = ktime_get();
490         wake_up(&svcpt->scp_waitq);
491 }
492
493 static void ptlrpc_server_nthreads_check(struct ptlrpc_service *svc,
494                                          struct ptlrpc_service_conf *conf)
495 {
496         struct ptlrpc_service_thr_conf *tc = &conf->psc_thr;
497         unsigned int init;
498         unsigned int total;
499         unsigned int nthrs;
500         int weight;
501
502         /*
503          * Common code for estimating & validating threads number.
504          * CPT affinity service could have percpt thread-pool instead
505          * of a global thread-pool, which means user might not always
506          * get the threads number they give it in conf::tc_nthrs_user
507          * even they did set. It's because we need to validate threads
508          * number for each CPT to guarantee each pool will have enough
509          * threads to keep the service healthy.
510          */
511         init = PTLRPC_NTHRS_INIT + (svc->srv_ops.so_hpreq_handler != NULL);
512         init = max_t(int, init, tc->tc_nthrs_init);
513
514         /*
515          * NB: please see comments in lustre_lnet.h for definition
516          * details of these members
517          */
518         LASSERT(tc->tc_nthrs_max != 0);
519
520         if (tc->tc_nthrs_user != 0) {
521                 /*
522                  * In case there is a reason to test a service with many
523                  * threads, we give a less strict check here, it can
524                  * be up to 8 * nthrs_max
525                  */
526                 total = min(tc->tc_nthrs_max * 8, tc->tc_nthrs_user);
527                 nthrs = total / svc->srv_ncpts;
528                 init  = max(init, nthrs);
529                 goto out;
530         }
531
532         total = tc->tc_nthrs_max;
533         if (tc->tc_nthrs_base == 0) {
534                 /*
535                  * don't care about base threads number per partition,
536                  * this is most for non-affinity service
537                  */
538                 nthrs = total / svc->srv_ncpts;
539                 goto out;
540         }
541
542         nthrs = tc->tc_nthrs_base;
543         if (svc->srv_ncpts == 1) {
544                 int     i;
545
546                 /*
547                  * NB: Increase the base number if it's single partition
548                  * and total number of cores/HTs is larger or equal to 4.
549                  * result will always < 2 * nthrs_base
550                  */
551                 weight = cfs_cpt_weight(svc->srv_cptable, CFS_CPT_ANY);
552                 for (i = 1; (weight >> (i + 1)) != 0 && /* >= 4 cores/HTs */
553                             (tc->tc_nthrs_base >> i) != 0; i++)
554                         nthrs += tc->tc_nthrs_base >> i;
555         }
556
557         if (tc->tc_thr_factor != 0) {
558                 int       factor = tc->tc_thr_factor;
559                 const int fade = 4;
560
561                 /*
562                  * User wants to increase number of threads with for
563                  * each CPU core/HT, most likely the factor is larger than
564                  * one thread/core because service threads are supposed to
565                  * be blocked by lock or wait for IO.
566                  */
567                 /*
568                  * Amdahl's law says that adding processors wouldn't give
569                  * a linear increasing of parallelism, so it's nonsense to
570                  * have too many threads no matter how many cores/HTs
571                  * there are.
572                  */
573                 preempt_disable();
574                 if (cpumask_weight
575                     (topology_sibling_cpumask(smp_processor_id())) > 1) {
576                         /* weight is # of HTs */
577                         /* depress thread factor for hyper-thread */
578                         factor = factor - (factor >> 1) + (factor >> 3);
579                 }
580                 preempt_enable();
581
582                 weight = cfs_cpt_weight(svc->srv_cptable, 0);
583
584                 for (; factor > 0 && weight > 0; factor--, weight -= fade)
585                         nthrs += min(weight, fade) * factor;
586         }
587
588         if (nthrs * svc->srv_ncpts > tc->tc_nthrs_max) {
589                 nthrs = max(tc->tc_nthrs_base,
590                             tc->tc_nthrs_max / svc->srv_ncpts);
591         }
592  out:
593         nthrs = max(nthrs, tc->tc_nthrs_init);
594         svc->srv_nthrs_cpt_limit = nthrs;
595         svc->srv_nthrs_cpt_init = init;
596
597         if (nthrs * svc->srv_ncpts > tc->tc_nthrs_max) {
598                 CDEBUG(D_OTHER,
599                        "%s: This service may have more threads (%d) than the given soft limit (%d)\n",
600                        svc->srv_name, nthrs * svc->srv_ncpts,
601                        tc->tc_nthrs_max);
602         }
603 }
604
605 /**
606  * Initialize percpt data for a service
607  */
608 static int ptlrpc_service_part_init(struct ptlrpc_service *svc,
609                                     struct ptlrpc_service_part *svcpt, int cpt)
610 {
611         struct ptlrpc_at_array *array;
612         int size;
613         int index;
614         int rc;
615
616         svcpt->scp_cpt = cpt;
617         INIT_LIST_HEAD(&svcpt->scp_threads);
618
619         /* rqbd and incoming request queue */
620         spin_lock_init(&svcpt->scp_lock);
621         mutex_init(&svcpt->scp_mutex);
622         INIT_LIST_HEAD(&svcpt->scp_rqbd_idle);
623         INIT_LIST_HEAD(&svcpt->scp_rqbd_posted);
624         INIT_LIST_HEAD(&svcpt->scp_req_incoming);
625         init_waitqueue_head(&svcpt->scp_waitq);
626         /* history request & rqbd list */
627         INIT_LIST_HEAD(&svcpt->scp_hist_reqs);
628         INIT_LIST_HEAD(&svcpt->scp_hist_rqbds);
629
630         /* acitve requests and hp requests */
631         spin_lock_init(&svcpt->scp_req_lock);
632
633         /* reply states */
634         spin_lock_init(&svcpt->scp_rep_lock);
635         INIT_LIST_HEAD(&svcpt->scp_rep_active);
636         INIT_LIST_HEAD(&svcpt->scp_rep_idle);
637         init_waitqueue_head(&svcpt->scp_rep_waitq);
638         atomic_set(&svcpt->scp_nreps_difficult, 0);
639
640         /* adaptive timeout */
641         spin_lock_init(&svcpt->scp_at_lock);
642         array = &svcpt->scp_at_array;
643
644         size = at_est2timeout(at_max);
645         array->paa_size     = size;
646         array->paa_count    = 0;
647         array->paa_deadline = -1;
648
649         /* allocate memory for scp_at_array (ptlrpc_at_array) */
650         OBD_CPT_ALLOC(array->paa_reqs_array,
651                       svc->srv_cptable, cpt, sizeof(struct list_head) * size);
652         if (array->paa_reqs_array == NULL)
653                 return -ENOMEM;
654
655         for (index = 0; index < size; index++)
656                 INIT_LIST_HEAD(&array->paa_reqs_array[index]);
657
658         OBD_CPT_ALLOC(array->paa_reqs_count,
659                       svc->srv_cptable, cpt, sizeof(__u32) * size);
660         if (array->paa_reqs_count == NULL)
661                 goto failed;
662
663         cfs_timer_setup(&svcpt->scp_at_timer, ptlrpc_at_timer,
664                         (unsigned long)svcpt, 0);
665
666         /*
667          * At SOW, service time should be quick; 10s seems generous. If client
668          * timeout is less than this, we'll be sending an early reply.
669          */
670         at_init(&svcpt->scp_at_estimate, 10, 0);
671
672         /* assign this before call ptlrpc_grow_req_bufs */
673         svcpt->scp_service = svc;
674         /* Now allocate the request buffers, but don't post them now */
675         rc = ptlrpc_grow_req_bufs(svcpt, 0);
676         /*
677          * We shouldn't be under memory pressure at startup, so
678          * fail if we can't allocate all our buffers at this time.
679          */
680         if (rc != 0)
681                 goto failed;
682
683         return 0;
684
685  failed:
686         if (array->paa_reqs_count != NULL) {
687                 OBD_FREE_PTR_ARRAY(array->paa_reqs_count, size);
688                 array->paa_reqs_count = NULL;
689         }
690
691         if (array->paa_reqs_array != NULL) {
692                 OBD_FREE_PTR_ARRAY(array->paa_reqs_array, array->paa_size);
693                 array->paa_reqs_array = NULL;
694         }
695
696         return -ENOMEM;
697 }
698
699 /**
700  * Initialize service on a given portal.
701  * This includes starting serving threads , allocating and posting rqbds and
702  * so on.
703  */
704 struct ptlrpc_service *ptlrpc_register_service(struct ptlrpc_service_conf *conf,
705                                                struct kset *parent,
706                                                struct dentry *debugfs_entry)
707 {
708         struct ptlrpc_service_cpt_conf *cconf = &conf->psc_cpt;
709         struct ptlrpc_service *service;
710         struct ptlrpc_service_part *svcpt;
711         struct cfs_cpt_table *cptable;
712         __u32 *cpts = NULL;
713         int ncpts;
714         int cpt;
715         int rc;
716         int i;
717
718         ENTRY;
719
720         LASSERT(conf->psc_buf.bc_nbufs > 0);
721         LASSERT(conf->psc_buf.bc_buf_size >=
722                 conf->psc_buf.bc_req_max_size + SPTLRPC_MAX_PAYLOAD);
723         LASSERT(conf->psc_thr.tc_ctx_tags != 0);
724
725         cptable = cconf->cc_cptable;
726         if (cptable == NULL)
727                 cptable = cfs_cpt_tab;
728
729         if (conf->psc_thr.tc_cpu_bind > 1) {
730                 CERROR("%s: Invalid cpu bind value %d, only 1 or 0 allowed\n",
731                        conf->psc_name, conf->psc_thr.tc_cpu_bind);
732                 RETURN(ERR_PTR(-EINVAL));
733         }
734
735         if (!cconf->cc_affinity) {
736                 ncpts = 1;
737         } else {
738                 ncpts = cfs_cpt_number(cptable);
739                 if (cconf->cc_pattern != NULL) {
740                         struct cfs_expr_list    *el;
741
742                         rc = cfs_expr_list_parse(cconf->cc_pattern,
743                                                  strlen(cconf->cc_pattern),
744                                                  0, ncpts - 1, &el);
745                         if (rc != 0) {
746                                 CERROR("%s: invalid CPT pattern string: %s",
747                                        conf->psc_name, cconf->cc_pattern);
748                                 RETURN(ERR_PTR(-EINVAL));
749                         }
750
751                         rc = cfs_expr_list_values(el, ncpts, &cpts);
752                         cfs_expr_list_free(el);
753                         if (rc <= 0) {
754                                 CERROR("%s: failed to parse CPT array %s: %d\n",
755                                        conf->psc_name, cconf->cc_pattern, rc);
756                                 if (cpts != NULL)
757                                         OBD_FREE_PTR_ARRAY(cpts, ncpts);
758                                 RETURN(ERR_PTR(rc < 0 ? rc : -EINVAL));
759                         }
760                         ncpts = rc;
761                 }
762         }
763
764         OBD_ALLOC(service, offsetof(struct ptlrpc_service, srv_parts[ncpts]));
765         if (service == NULL) {
766                 if (cpts != NULL)
767                         OBD_FREE_PTR_ARRAY(cpts, ncpts);
768                 RETURN(ERR_PTR(-ENOMEM));
769         }
770
771         service->srv_cptable            = cptable;
772         service->srv_cpts               = cpts;
773         service->srv_ncpts              = ncpts;
774         service->srv_cpt_bind           = conf->psc_thr.tc_cpu_bind;
775
776         service->srv_cpt_bits = 0; /* it's zero already, easy to read... */
777         while ((1 << service->srv_cpt_bits) < cfs_cpt_number(cptable))
778                 service->srv_cpt_bits++;
779
780         /* public members */
781         spin_lock_init(&service->srv_lock);
782         service->srv_name               = conf->psc_name;
783         service->srv_watchdog_factor    = conf->psc_watchdog_factor;
784         INIT_LIST_HEAD(&service->srv_list); /* for safty of cleanup */
785
786         /* buffer configuration */
787         service->srv_nbuf_per_group     = test_req_buffer_pressure ?
788                                           1 : conf->psc_buf.bc_nbufs;
789         /* do not limit max number of rqbds by default */
790         service->srv_nrqbds_max         = 0;
791
792         service->srv_max_req_size       = conf->psc_buf.bc_req_max_size +
793                                           SPTLRPC_MAX_PAYLOAD;
794         service->srv_buf_size           = conf->psc_buf.bc_buf_size;
795         service->srv_rep_portal         = conf->psc_buf.bc_rep_portal;
796         service->srv_req_portal         = conf->psc_buf.bc_req_portal;
797
798         /* With slab/alloc_pages buffer size will be rounded up to 2^n */
799         if (service->srv_buf_size & (service->srv_buf_size - 1)) {
800                 int round = size_roundup_power2(service->srv_buf_size);
801
802                 service->srv_buf_size = round;
803         }
804
805         /* Increase max reply size to next power of two */
806         service->srv_max_reply_size = 1;
807         while (service->srv_max_reply_size <
808                conf->psc_buf.bc_rep_max_size + SPTLRPC_MAX_PAYLOAD)
809                 service->srv_max_reply_size <<= 1;
810
811         service->srv_thread_name        = conf->psc_thr.tc_thr_name;
812         service->srv_ctx_tags           = conf->psc_thr.tc_ctx_tags;
813         service->srv_hpreq_ratio        = PTLRPC_SVC_HP_RATIO;
814         service->srv_ops                = conf->psc_ops;
815
816         for (i = 0; i < ncpts; i++) {
817                 if (!cconf->cc_affinity)
818                         cpt = CFS_CPT_ANY;
819                 else
820                         cpt = cpts != NULL ? cpts[i] : i;
821
822                 OBD_CPT_ALLOC(svcpt, cptable, cpt, sizeof(*svcpt));
823                 if (svcpt == NULL)
824                         GOTO(failed, rc = -ENOMEM);
825
826                 service->srv_parts[i] = svcpt;
827                 rc = ptlrpc_service_part_init(service, svcpt, cpt);
828                 if (rc != 0)
829                         GOTO(failed, rc);
830         }
831
832         ptlrpc_server_nthreads_check(service, conf);
833
834         rc = LNetSetLazyPortal(service->srv_req_portal);
835         LASSERT(rc == 0);
836
837         mutex_lock(&ptlrpc_all_services_mutex);
838         list_add(&service->srv_list, &ptlrpc_all_services);
839         mutex_unlock(&ptlrpc_all_services_mutex);
840
841         if (parent) {
842                 rc = ptlrpc_sysfs_register_service(parent, service);
843                 if (rc)
844                         GOTO(failed, rc);
845         }
846
847         if (debugfs_entry != NULL)
848                 ptlrpc_ldebugfs_register_service(debugfs_entry, service);
849
850         rc = ptlrpc_service_nrs_setup(service);
851         if (rc != 0)
852                 GOTO(failed, rc);
853
854         CDEBUG(D_NET, "%s: Started, listening on portal %d\n",
855                service->srv_name, service->srv_req_portal);
856
857         rc = ptlrpc_start_threads(service);
858         if (rc != 0) {
859                 CERROR("Failed to start threads for service %s: %d\n",
860                        service->srv_name, rc);
861                 GOTO(failed, rc);
862         }
863
864         RETURN(service);
865 failed:
866         ptlrpc_unregister_service(service);
867         RETURN(ERR_PTR(rc));
868 }
869 EXPORT_SYMBOL(ptlrpc_register_service);
870
871 /**
872  * to actually free the request, must be called without holding svc_lock.
873  * note it's caller's responsibility to unlink req->rq_list.
874  */
875 static void ptlrpc_server_free_request(struct ptlrpc_request *req)
876 {
877         LASSERT(atomic_read(&req->rq_refcount) == 0);
878         LASSERT(list_empty(&req->rq_timed_list));
879
880         /*
881          * DEBUG_REQ() assumes the reply state of a request with a valid
882          * ref will not be destroyed until that reference is dropped.
883          */
884         ptlrpc_req_drop_rs(req);
885
886         sptlrpc_svc_ctx_decref(req);
887
888         if (req != &req->rq_rqbd->rqbd_req) {
889                 /*
890                  * NB request buffers use an embedded
891                  * req if the incoming req unlinked the
892                  * MD; this isn't one of them!
893                  */
894                 ptlrpc_request_cache_free(req);
895         }
896 }
897
898 /**
899  * drop a reference count of the request. if it reaches 0, we either
900  * put it into history list, or free it immediately.
901  */
902 void ptlrpc_server_drop_request(struct ptlrpc_request *req)
903 {
904         struct ptlrpc_request_buffer_desc *rqbd = req->rq_rqbd;
905         struct ptlrpc_service_part        *svcpt = rqbd->rqbd_svcpt;
906         struct ptlrpc_service             *svc = svcpt->scp_service;
907         int                                refcount;
908         struct list_head                          *tmp;
909         struct list_head                          *nxt;
910
911         if (!atomic_dec_and_test(&req->rq_refcount))
912                 return;
913
914         if (req->rq_session.lc_state == LCS_ENTERED) {
915                 lu_context_exit(&req->rq_session);
916                 lu_context_fini(&req->rq_session);
917         }
918
919         if (req->rq_at_linked) {
920                 spin_lock(&svcpt->scp_at_lock);
921                 /*
922                  * recheck with lock, in case it's unlinked by
923                  * ptlrpc_at_check_timed()
924                  */
925                 if (likely(req->rq_at_linked))
926                         ptlrpc_at_remove_timed(req);
927                 spin_unlock(&svcpt->scp_at_lock);
928         }
929
930         LASSERT(list_empty(&req->rq_timed_list));
931
932         /* finalize request */
933         if (req->rq_export) {
934                 class_export_put(req->rq_export);
935                 req->rq_export = NULL;
936         }
937
938         spin_lock(&svcpt->scp_lock);
939
940         list_add(&req->rq_list, &rqbd->rqbd_reqs);
941
942         refcount = --(rqbd->rqbd_refcount);
943         if (refcount == 0) {
944                 /* request buffer is now idle: add to history */
945                 list_move_tail(&rqbd->rqbd_list, &svcpt->scp_hist_rqbds);
946                 svcpt->scp_hist_nrqbds++;
947
948                 /*
949                  * cull some history?
950                  * I expect only about 1 or 2 rqbds need to be recycled here
951                  */
952                 while (svcpt->scp_hist_nrqbds > svc->srv_hist_nrqbds_cpt_max) {
953                         rqbd = list_entry(svcpt->scp_hist_rqbds.next,
954                                           struct ptlrpc_request_buffer_desc,
955                                           rqbd_list);
956
957                         list_del(&rqbd->rqbd_list);
958                         svcpt->scp_hist_nrqbds--;
959
960                         /*
961                          * remove rqbd's reqs from svc's req history while
962                          * I've got the service lock
963                          */
964                         list_for_each(tmp, &rqbd->rqbd_reqs) {
965                                 req = list_entry(tmp, struct ptlrpc_request,
966                                                  rq_list);
967                                 /* Track the highest culled req seq */
968                                 if (req->rq_history_seq >
969                                     svcpt->scp_hist_seq_culled) {
970                                         svcpt->scp_hist_seq_culled =
971                                                 req->rq_history_seq;
972                                 }
973                                 list_del(&req->rq_history_list);
974                         }
975
976                         spin_unlock(&svcpt->scp_lock);
977
978                         list_for_each_safe(tmp, nxt, &rqbd->rqbd_reqs) {
979                                 req = list_entry(rqbd->rqbd_reqs.next,
980                                                  struct ptlrpc_request,
981                                                  rq_list);
982                                 list_del(&req->rq_list);
983                                 ptlrpc_server_free_request(req);
984                         }
985
986                         spin_lock(&svcpt->scp_lock);
987                         /*
988                          * now all reqs including the embedded req has been
989                          * disposed, schedule request buffer for re-use
990                          * or free it to drain some in excess.
991                          */
992                         LASSERT(atomic_read(&rqbd->rqbd_req.rq_refcount) == 0);
993                         if (svcpt->scp_nrqbds_posted >=
994                             svc->srv_nbuf_per_group ||
995                             (svc->srv_nrqbds_max != 0 &&
996                              svcpt->scp_nrqbds_total > svc->srv_nrqbds_max) ||
997                             test_req_buffer_pressure) {
998                                 /* like in ptlrpc_free_rqbd() */
999                                 svcpt->scp_nrqbds_total--;
1000                                 OBD_FREE_LARGE(rqbd->rqbd_buffer,
1001                                                svc->srv_buf_size);
1002                                 OBD_FREE_PTR(rqbd);
1003                         } else {
1004                                 list_add_tail(&rqbd->rqbd_list,
1005                                               &svcpt->scp_rqbd_idle);
1006                         }
1007                 }
1008
1009                 spin_unlock(&svcpt->scp_lock);
1010         } else if (req->rq_reply_state && req->rq_reply_state->rs_prealloc) {
1011                 /* If we are low on memory, we are not interested in history */
1012                 list_del(&req->rq_list);
1013                 list_del_init(&req->rq_history_list);
1014
1015                 /* Track the highest culled req seq */
1016                 if (req->rq_history_seq > svcpt->scp_hist_seq_culled)
1017                         svcpt->scp_hist_seq_culled = req->rq_history_seq;
1018
1019                 spin_unlock(&svcpt->scp_lock);
1020
1021                 ptlrpc_server_free_request(req);
1022         } else {
1023                 spin_unlock(&svcpt->scp_lock);
1024         }
1025 }
1026
1027 static void ptlrpc_add_exp_list_nolock(struct ptlrpc_request *req,
1028                                        struct obd_export *export, bool hp)
1029 {
1030         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
1031
1032         if (hp)
1033                 list_add(&req->rq_exp_list, &export->exp_hp_rpcs);
1034         else
1035                 list_add(&req->rq_exp_list, &export->exp_reg_rpcs);
1036         if (tag && export->exp_used_slots)
1037                 set_bit(tag - 1, export->exp_used_slots);
1038 }
1039
1040 static void ptlrpc_del_exp_list(struct ptlrpc_request *req)
1041 {
1042         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
1043
1044         spin_lock(&req->rq_export->exp_rpc_lock);
1045         list_del_init(&req->rq_exp_list);
1046         if (tag && !req->rq_obsolete && req->rq_export->exp_used_slots)
1047                 clear_bit(tag - 1, req->rq_export->exp_used_slots);
1048         spin_unlock(&req->rq_export->exp_rpc_lock);
1049 }
1050
1051 /** Change request export and move hp request from old export to new */
1052 void ptlrpc_request_change_export(struct ptlrpc_request *req,
1053                                   struct obd_export *export)
1054 {
1055         if (req->rq_export != NULL) {
1056                 LASSERT(!list_empty(&req->rq_exp_list));
1057                 /* remove rq_exp_list from last export */
1058                 ptlrpc_del_exp_list(req);
1059                 /* export has one reference already, so it's safe to
1060                  * add req to export queue here and get another
1061                  * reference for request later
1062                  */
1063                 spin_lock(&export->exp_rpc_lock);
1064                 ptlrpc_add_exp_list_nolock(req, export, req->rq_ops != NULL);
1065                 spin_unlock(&export->exp_rpc_lock);
1066
1067                 class_export_rpc_dec(req->rq_export);
1068                 class_export_put(req->rq_export);
1069         }
1070
1071         /* request takes one export refcount */
1072         req->rq_export = class_export_get(export);
1073         class_export_rpc_inc(export);
1074 }
1075
1076 /**
1077  * to finish a request: stop sending more early replies, and release
1078  * the request.
1079  */
1080 static void ptlrpc_server_finish_request(struct ptlrpc_service_part *svcpt,
1081                                          struct ptlrpc_request *req)
1082 {
1083         ptlrpc_server_hpreq_fini(req);
1084
1085         ptlrpc_server_drop_request(req);
1086 }
1087
1088 /**
1089  * to finish an active request: stop sending more early replies, and release
1090  * the request. should be called after we finished handling the request.
1091  */
1092 static void ptlrpc_server_finish_active_request(
1093                                         struct ptlrpc_service_part *svcpt,
1094                                         struct ptlrpc_request *req)
1095 {
1096         spin_lock(&svcpt->scp_req_lock);
1097         ptlrpc_nrs_req_stop_nolock(req);
1098         svcpt->scp_nreqs_active--;
1099         if (req->rq_hp)
1100                 svcpt->scp_nhreqs_active--;
1101         spin_unlock(&svcpt->scp_req_lock);
1102
1103         ptlrpc_nrs_req_finalize(req);
1104
1105         if (req->rq_export != NULL)
1106                 class_export_rpc_dec(req->rq_export);
1107
1108         ptlrpc_server_finish_request(svcpt, req);
1109 }
1110
1111 /**
1112  * This function makes sure dead exports are evicted in a timely manner.
1113  * This function is only called when some export receives a message (i.e.,
1114  * the network is up.)
1115  */
1116 void ptlrpc_update_export_timer(struct obd_export *exp, time64_t extra_delay)
1117 {
1118         struct obd_export *oldest_exp;
1119         time64_t oldest_time, new_time;
1120
1121         ENTRY;
1122
1123         LASSERT(exp);
1124
1125         /*
1126          * Compensate for slow machines, etc, by faking our request time
1127          * into the future.  Although this can break the strict time-ordering
1128          * of the list, we can be really lazy here - we don't have to evict
1129          * at the exact right moment.  Eventually, all silent exports
1130          * will make it to the top of the list.
1131          */
1132
1133         /* Do not pay attention on 1sec or smaller renewals. */
1134         new_time = ktime_get_real_seconds() + extra_delay;
1135         if (exp->exp_last_request_time + 1 /*second */ >= new_time)
1136                 RETURN_EXIT;
1137
1138         exp->exp_last_request_time = new_time;
1139
1140         /*
1141          * exports may get disconnected from the chain even though the
1142          * export has references, so we must keep the spin lock while
1143          * manipulating the lists
1144          */
1145         spin_lock(&exp->exp_obd->obd_dev_lock);
1146
1147         if (list_empty(&exp->exp_obd_chain_timed)) {
1148                 /* this one is not timed */
1149                 spin_unlock(&exp->exp_obd->obd_dev_lock);
1150                 RETURN_EXIT;
1151         }
1152
1153         list_move_tail(&exp->exp_obd_chain_timed,
1154                        &exp->exp_obd->obd_exports_timed);
1155
1156         oldest_exp = list_entry(exp->exp_obd->obd_exports_timed.next,
1157                                 struct obd_export, exp_obd_chain_timed);
1158         oldest_time = oldest_exp->exp_last_request_time;
1159         spin_unlock(&exp->exp_obd->obd_dev_lock);
1160
1161         if (exp->exp_obd->obd_recovering) {
1162                 /* be nice to everyone during recovery */
1163                 EXIT;
1164                 return;
1165         }
1166
1167         /* Note - racing to start/reset the obd_eviction timer is safe */
1168         if (exp->exp_obd->obd_eviction_timer == 0) {
1169                 /* Check if the oldest entry is expired. */
1170                 if (ktime_get_real_seconds() >
1171                     oldest_time + PING_EVICT_TIMEOUT + extra_delay) {
1172                         /*
1173                          * We need a second timer, in case the net was down and
1174                          * it just came back. Since the pinger may skip every
1175                          * other PING_INTERVAL (see note in ptlrpc_pinger_main),
1176                          * we better wait for 3.
1177                          */
1178                         exp->exp_obd->obd_eviction_timer =
1179                                 ktime_get_real_seconds() + 3 * PING_INTERVAL;
1180                         CDEBUG(D_HA, "%s: Think about evicting %s from %lld\n",
1181                                exp->exp_obd->obd_name,
1182                                obd_export_nid2str(oldest_exp), oldest_time);
1183                 }
1184         } else {
1185                 if (ktime_get_real_seconds() >
1186                     (exp->exp_obd->obd_eviction_timer + extra_delay)) {
1187                         /*
1188                          * The evictor won't evict anyone who we've heard from
1189                          * recently, so we don't have to check before we start
1190                          * it.
1191                          */
1192                         if (!ping_evictor_wake(exp))
1193                                 exp->exp_obd->obd_eviction_timer = 0;
1194                 }
1195         }
1196
1197         EXIT;
1198 }
1199
1200 /**
1201  * Sanity check request \a req.
1202  * Return 0 if all is ok, error code otherwise.
1203  */
1204 static int ptlrpc_check_req(struct ptlrpc_request *req)
1205 {
1206         struct obd_device *obd = req->rq_export->exp_obd;
1207         int rc = 0;
1208
1209         if (unlikely(lustre_msg_get_conn_cnt(req->rq_reqmsg) <
1210                      req->rq_export->exp_conn_cnt)) {
1211                 DEBUG_REQ(D_RPCTRACE, req,
1212                           "DROPPING req from old connection %d < %d",
1213                           lustre_msg_get_conn_cnt(req->rq_reqmsg),
1214                           req->rq_export->exp_conn_cnt);
1215                 return -EEXIST;
1216         }
1217         if (unlikely(obd == NULL || obd->obd_fail)) {
1218                 /*
1219                  * Failing over, don't handle any more reqs,
1220                  * send error response instead.
1221                  */
1222                 CDEBUG(D_RPCTRACE, "Dropping req %p for failed obd %s\n",
1223                         req, (obd != NULL) ? obd->obd_name : "unknown");
1224                 rc = -ENODEV;
1225         } else if (lustre_msg_get_flags(req->rq_reqmsg) &
1226                    (MSG_REPLAY | MSG_REQ_REPLAY_DONE) &&
1227                    !obd->obd_recovering) {
1228                 DEBUG_REQ(D_ERROR, req,
1229                           "Invalid replay without recovery");
1230                 class_fail_export(req->rq_export);
1231                 rc = -ENODEV;
1232         } else if (lustre_msg_get_transno(req->rq_reqmsg) != 0 &&
1233                    !obd->obd_recovering) {
1234                 DEBUG_REQ(D_ERROR, req,
1235                           "Invalid req with transno %llu without recovery",
1236                           lustre_msg_get_transno(req->rq_reqmsg));
1237                 class_fail_export(req->rq_export);
1238                 rc = -ENODEV;
1239         }
1240
1241         if (unlikely(rc < 0)) {
1242                 req->rq_status = rc;
1243                 ptlrpc_error(req);
1244         }
1245         return rc;
1246 }
1247
1248 static void ptlrpc_at_set_timer(struct ptlrpc_service_part *svcpt)
1249 {
1250         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1251         time64_t next;
1252
1253         if (array->paa_count == 0) {
1254                 del_timer(&svcpt->scp_at_timer);
1255                 return;
1256         }
1257
1258         /* Set timer for closest deadline */
1259         next = array->paa_deadline - ktime_get_real_seconds() -
1260                at_early_margin;
1261         if (next <= 0) {
1262                 ptlrpc_at_timer(cfs_timer_cb_arg(svcpt, scp_at_timer));
1263         } else {
1264                 mod_timer(&svcpt->scp_at_timer,
1265                           jiffies + nsecs_to_jiffies(next * NSEC_PER_SEC));
1266                 CDEBUG(D_INFO, "armed %s at %+llds\n",
1267                        svcpt->scp_service->srv_name, next);
1268         }
1269 }
1270
1271 /* Add rpc to early reply check list */
1272 static int ptlrpc_at_add_timed(struct ptlrpc_request *req)
1273 {
1274         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1275         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1276         struct ptlrpc_request *rq = NULL;
1277         __u32 index;
1278
1279         if (AT_OFF)
1280                 return(0);
1281
1282         if (req->rq_no_reply)
1283                 return 0;
1284
1285         if ((lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT) == 0)
1286                 return(-ENOSYS);
1287
1288         spin_lock(&svcpt->scp_at_lock);
1289         LASSERT(list_empty(&req->rq_timed_list));
1290
1291         div_u64_rem(req->rq_deadline, array->paa_size, &index);
1292         if (array->paa_reqs_count[index] > 0) {
1293                 /*
1294                  * latest rpcs will have the latest deadlines in the list,
1295                  * so search backward.
1296                  */
1297                 list_for_each_entry_reverse(rq, &array->paa_reqs_array[index],
1298                                             rq_timed_list) {
1299                         if (req->rq_deadline >= rq->rq_deadline) {
1300                                 list_add(&req->rq_timed_list,
1301                                          &rq->rq_timed_list);
1302                                 break;
1303                         }
1304                 }
1305         }
1306
1307         /* Add the request at the head of the list */
1308         if (list_empty(&req->rq_timed_list))
1309                 list_add(&req->rq_timed_list, &array->paa_reqs_array[index]);
1310
1311         spin_lock(&req->rq_lock);
1312         req->rq_at_linked = 1;
1313         spin_unlock(&req->rq_lock);
1314         req->rq_at_index = index;
1315         array->paa_reqs_count[index]++;
1316         array->paa_count++;
1317         if (array->paa_count == 1 || array->paa_deadline > req->rq_deadline) {
1318                 array->paa_deadline = req->rq_deadline;
1319                 ptlrpc_at_set_timer(svcpt);
1320         }
1321         spin_unlock(&svcpt->scp_at_lock);
1322
1323         return 0;
1324 }
1325
1326 static void ptlrpc_at_remove_timed(struct ptlrpc_request *req)
1327 {
1328         struct ptlrpc_at_array *array;
1329
1330         array = &req->rq_rqbd->rqbd_svcpt->scp_at_array;
1331
1332         /* NB: must call with hold svcpt::scp_at_lock */
1333         LASSERT(!list_empty(&req->rq_timed_list));
1334         list_del_init(&req->rq_timed_list);
1335
1336         spin_lock(&req->rq_lock);
1337         req->rq_at_linked = 0;
1338         spin_unlock(&req->rq_lock);
1339
1340         array->paa_reqs_count[req->rq_at_index]--;
1341         array->paa_count--;
1342 }
1343
1344 /*
1345  * Attempt to extend the request deadline by sending an early reply to the
1346  * client.
1347  */
1348 static int ptlrpc_at_send_early_reply(struct ptlrpc_request *req)
1349 {
1350         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1351         struct ptlrpc_request *reqcopy;
1352         struct lustre_msg *reqmsg;
1353         timeout_t olddl = req->rq_deadline - ktime_get_real_seconds();
1354         time64_t newdl;
1355         int rc;
1356
1357         ENTRY;
1358
1359         if (CFS_FAIL_CHECK(OBD_FAIL_TGT_REPLAY_RECONNECT)) {
1360                 /* don't send early reply */
1361                 RETURN(1);
1362         }
1363
1364         /*
1365          * deadline is when the client expects us to reply, margin is the
1366          * difference between clients' and servers' expectations
1367          */
1368         DEBUG_REQ(D_ADAPTTO, req,
1369                   "%ssending early reply (deadline %+ds, margin %+ds) for %d+%d",
1370                   AT_OFF ? "AT off - not " : "",
1371                   olddl, olddl - at_get(&svcpt->scp_at_estimate),
1372                   at_get(&svcpt->scp_at_estimate), at_extra);
1373
1374         if (AT_OFF)
1375                 RETURN(0);
1376
1377         if (olddl < 0) {
1378                 /* below message is checked in replay-ost-single.sh test_9 */
1379                 DEBUG_REQ(D_WARNING, req,
1380                           "Already past deadline (%+ds), not sending early reply. Consider increasing at_early_margin (%d)?",
1381                           olddl, at_early_margin);
1382
1383                 /* Return an error so we're not re-added to the timed list. */
1384                 RETURN(-ETIMEDOUT);
1385         }
1386
1387         if ((lustre_msghdr_get_flags(req->rq_reqmsg) &
1388              MSGHDR_AT_SUPPORT) == 0) {
1389                 DEBUG_REQ(D_INFO, req,
1390                           "Wanted to ask client for more time, but no AT support");
1391                 RETURN(-ENOSYS);
1392         }
1393
1394         if (req->rq_export &&
1395             lustre_msg_get_flags(req->rq_reqmsg) &
1396             (MSG_REPLAY | MSG_REQ_REPLAY_DONE | MSG_LOCK_REPLAY_DONE)) {
1397                 struct obd_device *obd_exp = req->rq_export->exp_obd;
1398
1399                 /*
1400                  * During recovery, we don't want to send too many early
1401                  * replies, but on the other hand we want to make sure the
1402                  * client has enough time to resend if the rpc is lost. So
1403                  * during the recovery period send at least 4 early replies,
1404                  * spacing them every at_extra if we can. at_estimate should
1405                  * always equal this fixed value during recovery.
1406                  */
1407
1408                 /*
1409                  * Don't account request processing time into AT history
1410                  * during recovery, it is not service time we need but
1411                  * includes also waiting time for recovering clients
1412                  */
1413                 newdl = min_t(time64_t, at_extra,
1414                               obd_exp->obd_recovery_timeout / 4) +
1415                         ktime_get_real_seconds();
1416         } else {
1417                 /*
1418                  * We want to extend the request deadline by at_extra seconds,
1419                  * so we set our service estimate to reflect how much time has
1420                  * passed since this request arrived plus an additional
1421                  * at_extra seconds. The client will calculate the new deadline
1422                  * based on this service estimate (plus some additional time to
1423                  * account for network latency). See ptlrpc_at_recv_early_reply
1424                  */
1425                 at_measured(&svcpt->scp_at_estimate, at_extra +
1426                             ktime_get_real_seconds() -
1427                             req->rq_arrival_time.tv_sec);
1428                 newdl = req->rq_arrival_time.tv_sec +
1429                         at_get(&svcpt->scp_at_estimate);
1430         }
1431
1432         /*
1433          * Check to see if we've actually increased the deadline -
1434          * we may be past adaptive_max
1435          */
1436         if (req->rq_deadline >= newdl) {
1437                 DEBUG_REQ(D_WARNING, req,
1438                           "Could not add any time (%d/%lld), not sending early reply",
1439                           olddl, newdl - ktime_get_real_seconds());
1440                 RETURN(-ETIMEDOUT);
1441         }
1442
1443         reqcopy = ptlrpc_request_cache_alloc(GFP_NOFS);
1444         if (reqcopy == NULL)
1445                 RETURN(-ENOMEM);
1446         OBD_ALLOC_LARGE(reqmsg, req->rq_reqlen);
1447         if (!reqmsg)
1448                 GOTO(out_free, rc = -ENOMEM);
1449
1450         *reqcopy = *req;
1451         reqcopy->rq_reply_state = NULL;
1452         reqcopy->rq_rep_swab_mask = 0;
1453         reqcopy->rq_pack_bulk = 0;
1454         reqcopy->rq_pack_udesc = 0;
1455         reqcopy->rq_packed_final = 0;
1456         sptlrpc_svc_ctx_addref(reqcopy);
1457         /* We only need the reqmsg for the magic */
1458         reqcopy->rq_reqmsg = reqmsg;
1459         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
1460
1461         /*
1462          * tgt_brw_read() and tgt_brw_write() may have decided not to reply.
1463          * Without this check, we would fail the rq_no_reply assertion in
1464          * ptlrpc_send_reply().
1465          */
1466         if (reqcopy->rq_no_reply)
1467                 GOTO(out, rc = -ETIMEDOUT);
1468
1469         LASSERT(atomic_read(&req->rq_refcount));
1470         /* if it is last refcount then early reply isn't needed */
1471         if (atomic_read(&req->rq_refcount) == 1) {
1472                 DEBUG_REQ(D_ADAPTTO, reqcopy,
1473                           "Normal reply already sent, abort early reply");
1474                 GOTO(out, rc = -EINVAL);
1475         }
1476
1477         /* Connection ref */
1478         reqcopy->rq_export = class_conn2export(
1479                         lustre_msg_get_handle(reqcopy->rq_reqmsg));
1480         if (reqcopy->rq_export == NULL)
1481                 GOTO(out, rc = -ENODEV);
1482
1483         /* RPC ref */
1484         class_export_rpc_inc(reqcopy->rq_export);
1485         if (reqcopy->rq_export->exp_obd &&
1486             reqcopy->rq_export->exp_obd->obd_fail)
1487                 GOTO(out_put, rc = -ENODEV);
1488
1489         rc = lustre_pack_reply_flags(reqcopy, 1, NULL, NULL, LPRFL_EARLY_REPLY);
1490         if (rc)
1491                 GOTO(out_put, rc);
1492
1493         rc = ptlrpc_send_reply(reqcopy, PTLRPC_REPLY_EARLY);
1494
1495         if (!rc) {
1496                 /* Adjust our own deadline to what we told the client */
1497                 req->rq_deadline = newdl;
1498                 req->rq_early_count++; /* number sent, server side */
1499         } else {
1500                 DEBUG_REQ(D_ERROR, req, "Early reply send failed: rc = %d", rc);
1501         }
1502
1503         /*
1504          * Free the (early) reply state from lustre_pack_reply.
1505          * (ptlrpc_send_reply takes it's own rs ref, so this is safe here)
1506          */
1507         ptlrpc_req_drop_rs(reqcopy);
1508
1509 out_put:
1510         class_export_rpc_dec(reqcopy->rq_export);
1511         class_export_put(reqcopy->rq_export);
1512 out:
1513         sptlrpc_svc_ctx_decref(reqcopy);
1514         OBD_FREE_LARGE(reqmsg, req->rq_reqlen);
1515 out_free:
1516         ptlrpc_request_cache_free(reqcopy);
1517         RETURN(rc);
1518 }
1519
1520 /*
1521  * Send early replies to everybody expiring within at_early_margin
1522  * asking for at_extra time
1523  */
1524 static int ptlrpc_at_check_timed(struct ptlrpc_service_part *svcpt)
1525 {
1526         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1527         struct ptlrpc_request *rq, *n;
1528         LIST_HEAD(work_list);
1529         __u32 index, count;
1530         time64_t deadline;
1531         time64_t now = ktime_get_real_seconds();
1532         s64 delay_ms;
1533         int first, counter = 0;
1534
1535         ENTRY;
1536         spin_lock(&svcpt->scp_at_lock);
1537         if (svcpt->scp_at_check == 0) {
1538                 spin_unlock(&svcpt->scp_at_lock);
1539                 RETURN(0);
1540         }
1541         delay_ms = ktime_ms_delta(ktime_get(), svcpt->scp_at_checktime);
1542         svcpt->scp_at_check = 0;
1543
1544         if (array->paa_count == 0) {
1545                 spin_unlock(&svcpt->scp_at_lock);
1546                 RETURN(0);
1547         }
1548
1549         /* The timer went off, but maybe the nearest rpc already completed. */
1550         first = array->paa_deadline - now;
1551         if (first > at_early_margin) {
1552                 /* We've still got plenty of time.  Reset the timer. */
1553                 ptlrpc_at_set_timer(svcpt);
1554                 spin_unlock(&svcpt->scp_at_lock);
1555                 RETURN(0);
1556         }
1557
1558         /*
1559          * We're close to a timeout, and we don't know how much longer the
1560          * server will take. Send early replies to everyone expiring soon.
1561          */
1562         deadline = -1;
1563         div_u64_rem(array->paa_deadline, array->paa_size, &index);
1564         count = array->paa_count;
1565         while (count > 0) {
1566                 count -= array->paa_reqs_count[index];
1567                 list_for_each_entry_safe(rq, n,
1568                                          &array->paa_reqs_array[index],
1569                                          rq_timed_list) {
1570                         if (rq->rq_deadline > now + at_early_margin) {
1571                                 /* update the earliest deadline */
1572                                 if (deadline == -1 ||
1573                                     rq->rq_deadline < deadline)
1574                                         deadline = rq->rq_deadline;
1575                                 break;
1576                         }
1577
1578                         /**
1579                          * ptlrpc_server_drop_request() may drop
1580                          * refcount to 0 already. Let's check this and
1581                          * don't add entry to work_list
1582                          */
1583                         if (likely(atomic_inc_not_zero(&rq->rq_refcount))) {
1584                                 ptlrpc_at_remove_timed(rq);
1585                                 list_add(&rq->rq_timed_list, &work_list);
1586                         } else {
1587                                 ptlrpc_at_remove_timed(rq);
1588                         }
1589
1590                         counter++;
1591                 }
1592
1593                 if (++index >= array->paa_size)
1594                         index = 0;
1595         }
1596         array->paa_deadline = deadline;
1597         /* we have a new earliest deadline, restart the timer */
1598         ptlrpc_at_set_timer(svcpt);
1599
1600         spin_unlock(&svcpt->scp_at_lock);
1601
1602         CDEBUG(D_ADAPTTO,
1603                "timeout in %+ds, asking for %d secs on %d early replies\n",
1604                first, at_extra, counter);
1605         if (first < 0) {
1606                 /*
1607                  * We're already past request deadlines before we even get a
1608                  * chance to send early replies
1609                  */
1610                 LCONSOLE_WARN("%s: This server is not able to keep up with request traffic (cpu-bound).\n",
1611                               svcpt->scp_service->srv_name);
1612                 CWARN("earlyQ=%d reqQ=%d recA=%d, svcEst=%d, delay=%lldms\n",
1613                       counter, svcpt->scp_nreqs_incoming,
1614                       svcpt->scp_nreqs_active,
1615                       at_get(&svcpt->scp_at_estimate), delay_ms);
1616         }
1617
1618         /*
1619          * we took additional refcount so entries can't be deleted from list, no
1620          * locking is needed
1621          */
1622         while (!list_empty(&work_list)) {
1623                 rq = list_entry(work_list.next, struct ptlrpc_request,
1624                                 rq_timed_list);
1625                 list_del_init(&rq->rq_timed_list);
1626
1627                 if (ptlrpc_at_send_early_reply(rq) == 0)
1628                         ptlrpc_at_add_timed(rq);
1629
1630                 ptlrpc_server_drop_request(rq);
1631         }
1632
1633         RETURN(1); /* return "did_something" for liblustre */
1634 }
1635
1636 /*
1637  * Check if we are already handling earlier incarnation of this request.
1638  * Called under &req->rq_export->exp_rpc_lock locked
1639  */
1640 static struct ptlrpc_request*
1641 ptlrpc_server_check_resend_in_progress(struct ptlrpc_request *req)
1642 {
1643         struct ptlrpc_request *tmp = NULL;
1644
1645         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) ||
1646             (atomic_read(&req->rq_export->exp_rpc_count) == 0))
1647                 return NULL;
1648
1649         /*
1650          * bulk request are aborted upon reconnect, don't try to
1651          * find a match
1652          */
1653         if (req->rq_bulk_write || req->rq_bulk_read)
1654                 return NULL;
1655
1656         /*
1657          * This list should not be longer than max_requests in
1658          * flights on the client, so it is not all that long.
1659          * Also we only hit this codepath in case of a resent
1660          * request which makes it even more rarely hit
1661          */
1662         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs,
1663                                 rq_exp_list) {
1664                 /* Found duplicate one */
1665                 if (tmp->rq_xid == req->rq_xid)
1666                         goto found;
1667         }
1668         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs,
1669                                 rq_exp_list) {
1670                 /* Found duplicate one */
1671                 if (tmp->rq_xid == req->rq_xid)
1672                         goto found;
1673         }
1674         return NULL;
1675
1676 found:
1677         DEBUG_REQ(D_HA, req, "Found duplicate req in processing");
1678         DEBUG_REQ(D_HA, tmp, "Request being processed");
1679         return tmp;
1680 }
1681
1682 #ifdef HAVE_SERVER_SUPPORT
1683 static void ptlrpc_server_mark_obsolete(struct ptlrpc_request *req)
1684 {
1685         req->rq_obsolete = 1;
1686 }
1687
1688 static void
1689 ptlrpc_server_mark_in_progress_obsolete(struct ptlrpc_request *req)
1690 {
1691         struct ptlrpc_request   *tmp = NULL;
1692         __u16                   tag;
1693
1694         if (!tgt_is_increasing_xid_client(req->rq_export) ||
1695             req->rq_export->exp_used_slots == NULL)
1696                 return;
1697
1698         tag = lustre_msg_get_tag(req->rq_reqmsg);
1699         if (tag == 0)
1700                 return;
1701
1702         if (!test_bit(tag - 1, req->rq_export->exp_used_slots))
1703                 return;
1704
1705         /* This list should not be longer than max_requests in
1706          * flights on the client, so it is not all that long.
1707          * Also we only hit this codepath in case of a resent
1708          * request which makes it even more rarely hit */
1709         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs, rq_exp_list) {
1710                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1711                     req->rq_xid > tmp->rq_xid)
1712                         ptlrpc_server_mark_obsolete(tmp);
1713
1714         }
1715         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs, rq_exp_list) {
1716                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1717                     req->rq_xid > tmp->rq_xid)
1718                         ptlrpc_server_mark_obsolete(tmp);
1719         }
1720 }
1721 #endif
1722
1723 /**
1724  * Check if a request should be assigned with a high priority.
1725  *
1726  * \retval      < 0: error occurred
1727  *                0: normal RPC request
1728  *               +1: high priority request
1729  */
1730 static int ptlrpc_server_hpreq_init(struct ptlrpc_service_part *svcpt,
1731                                     struct ptlrpc_request *req)
1732 {
1733         int rc = 0;
1734
1735         ENTRY;
1736         if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL) {
1737                 rc = svcpt->scp_service->srv_ops.so_hpreq_handler(req);
1738                 if (rc < 0)
1739                         RETURN(rc);
1740
1741                 LASSERT(rc == 0);
1742         }
1743
1744         if (req->rq_export != NULL && req->rq_ops != NULL) {
1745                 /*
1746                  * Perform request specific check. We should do this
1747                  * check before the request is added into exp_hp_rpcs
1748                  * list otherwise it may hit swab race at LU-1044.
1749                  */
1750                 if (req->rq_ops->hpreq_check != NULL) {
1751                         rc = req->rq_ops->hpreq_check(req);
1752                         if (rc == -ESTALE) {
1753                                 req->rq_status = rc;
1754                                 ptlrpc_error(req);
1755                         }
1756                         /*
1757                          * can only return error,
1758                          * 0 for normal request,
1759                          * or 1 for high priority request
1760                          */
1761                         LASSERT(rc <= 1);
1762                 }
1763         }
1764
1765         RETURN(rc);
1766 }
1767
1768 /** Remove the request from the export list. */
1769 static void ptlrpc_server_hpreq_fini(struct ptlrpc_request *req)
1770 {
1771         ENTRY;
1772         if (req->rq_export) {
1773                 /*
1774                  * refresh lock timeout again so that client has more
1775                  * room to send lock cancel RPC.
1776                  */
1777                 if (req->rq_ops && req->rq_ops->hpreq_fini)
1778                         req->rq_ops->hpreq_fini(req);
1779
1780                 ptlrpc_del_exp_list(req);
1781         }
1782         EXIT;
1783 }
1784
1785 static int ptlrpc_hpreq_check(struct ptlrpc_request *req)
1786 {
1787         return 1;
1788 }
1789
1790 static struct ptlrpc_hpreq_ops ptlrpc_hpreq_common = {
1791         .hpreq_check       = ptlrpc_hpreq_check,
1792 };
1793
1794 /* Hi-Priority RPC check by RPC operation code. */
1795 int ptlrpc_hpreq_handler(struct ptlrpc_request *req)
1796 {
1797         int opc = lustre_msg_get_opc(req->rq_reqmsg);
1798
1799         /*
1800          * Check for export to let only reconnects for not yet evicted
1801          * export to become a HP rpc.
1802          */
1803         if ((req->rq_export != NULL) &&
1804             (opc == OBD_PING || opc == MDS_CONNECT || opc == OST_CONNECT))
1805                 req->rq_ops = &ptlrpc_hpreq_common;
1806
1807         return 0;
1808 }
1809 EXPORT_SYMBOL(ptlrpc_hpreq_handler);
1810
1811 static int ptlrpc_server_request_add(struct ptlrpc_service_part *svcpt,
1812                                      struct ptlrpc_request *req)
1813 {
1814         int rc;
1815         bool hp;
1816         struct ptlrpc_request *orig;
1817
1818         ENTRY;
1819
1820         rc = ptlrpc_server_hpreq_init(svcpt, req);
1821         if (rc < 0)
1822                 RETURN(rc);
1823
1824         hp = rc > 0;
1825         ptlrpc_nrs_req_initialize(svcpt, req, hp);
1826
1827         while (req->rq_export != NULL) {
1828                 struct obd_export *exp = req->rq_export;
1829
1830                 /*
1831                  * do search for duplicated xid and the adding to the list
1832                  * atomically
1833                  */
1834                 spin_lock_bh(&exp->exp_rpc_lock);
1835 #ifdef HAVE_SERVER_SUPPORT
1836                 ptlrpc_server_mark_in_progress_obsolete(req);
1837 #endif
1838                 orig = ptlrpc_server_check_resend_in_progress(req);
1839                 if (orig && OBD_FAIL_PRECHECK(OBD_FAIL_PTLRPC_RESEND_RACE)) {
1840                         spin_unlock_bh(&exp->exp_rpc_lock);
1841
1842                         OBD_RACE(OBD_FAIL_PTLRPC_RESEND_RACE);
1843                         msleep(4 * MSEC_PER_SEC);
1844                         continue;
1845                 }
1846
1847                 if (orig && likely(atomic_inc_not_zero(&orig->rq_refcount))) {
1848                         bool linked;
1849
1850                         spin_unlock_bh(&exp->exp_rpc_lock);
1851
1852                         /*
1853                          * When the client resend request and the server has
1854                          * the previous copy of it, we need to update deadlines,
1855                          * to be sure that the client and the server have equal
1856                          *  request deadlines.
1857                          */
1858
1859                         spin_lock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1860                         linked = orig->rq_at_linked;
1861                         if (likely(linked))
1862                                 ptlrpc_at_remove_timed(orig);
1863                         spin_unlock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1864                         orig->rq_deadline = req->rq_deadline;
1865                         if (likely(linked))
1866                                 ptlrpc_at_add_timed(orig);
1867                         ptlrpc_server_drop_request(orig);
1868                         ptlrpc_nrs_req_finalize(req);
1869
1870                         /* don't mark slot unused for resend in progress */
1871                         req->rq_obsolete = 1;
1872
1873                         RETURN(-EBUSY);
1874                 }
1875
1876                 ptlrpc_add_exp_list_nolock(req, exp, hp || req->rq_ops != NULL);
1877
1878                 spin_unlock_bh(&exp->exp_rpc_lock);
1879                 break;
1880         }
1881
1882         /*
1883          * the current thread is not the processing thread for this request
1884          * since that, but request is in exp_hp_list and can be find there.
1885          * Remove all relations between request and old thread.
1886          */
1887         req->rq_svc_thread->t_env->le_ses = NULL;
1888         req->rq_svc_thread = NULL;
1889         req->rq_session.lc_thread = NULL;
1890
1891         ptlrpc_nrs_req_add(svcpt, req, hp);
1892
1893         RETURN(0);
1894 }
1895
1896 /**
1897  * Allow to handle high priority request
1898  * User can call it w/o any lock but need to hold
1899  * ptlrpc_service_part::scp_req_lock to get reliable result
1900  */
1901 static bool ptlrpc_server_allow_high(struct ptlrpc_service_part *svcpt,
1902                                      bool force)
1903 {
1904         int running = svcpt->scp_nthrs_running;
1905
1906         if (!nrs_svcpt_has_hp(svcpt))
1907                 return false;
1908
1909         if (force)
1910                 return true;
1911
1912         if (ptlrpc_nrs_req_throttling_nolock(svcpt, true))
1913                 return false;
1914
1915         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1916                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1917                 /* leave just 1 thread for normal RPCs */
1918                 running = PTLRPC_NTHRS_INIT;
1919                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1920                         running += 1;
1921         }
1922
1923         if (svcpt->scp_nreqs_active >= running - 1)
1924                 return false;
1925
1926         if (svcpt->scp_nhreqs_active == 0)
1927                 return true;
1928
1929         return !ptlrpc_nrs_req_pending_nolock(svcpt, false) ||
1930                svcpt->scp_hreq_count < svcpt->scp_service->srv_hpreq_ratio;
1931 }
1932
1933 static bool ptlrpc_server_high_pending(struct ptlrpc_service_part *svcpt,
1934                                        bool force)
1935 {
1936         return ptlrpc_server_allow_high(svcpt, force) &&
1937                ptlrpc_nrs_req_pending_nolock(svcpt, true);
1938 }
1939
1940 /**
1941  * Only allow normal priority requests on a service that has a high-priority
1942  * queue if forced (i.e. cleanup), if there are other high priority requests
1943  * already being processed (i.e. those threads can service more high-priority
1944  * requests), or if there are enough idle threads that a later thread can do
1945  * a high priority request.
1946  * User can call it w/o any lock but need to hold
1947  * ptlrpc_service_part::scp_req_lock to get reliable result
1948  */
1949 static bool ptlrpc_server_allow_normal(struct ptlrpc_service_part *svcpt,
1950                                        bool force)
1951 {
1952         int running = svcpt->scp_nthrs_running;
1953
1954         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1955                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1956                 /* leave just 1 thread for normal RPCs */
1957                 running = PTLRPC_NTHRS_INIT;
1958                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1959                         running += 1;
1960         }
1961
1962         if (force)
1963                 return true;
1964
1965         if (ptlrpc_nrs_req_throttling_nolock(svcpt, false))
1966                 return false;
1967
1968         if (svcpt->scp_nreqs_active < running - 2)
1969                 return true;
1970
1971         if (svcpt->scp_nreqs_active >= running - 1)
1972                 return false;
1973
1974         return svcpt->scp_nhreqs_active > 0 || !nrs_svcpt_has_hp(svcpt);
1975 }
1976
1977 static bool ptlrpc_server_normal_pending(struct ptlrpc_service_part *svcpt,
1978                                          bool force)
1979 {
1980         return ptlrpc_server_allow_normal(svcpt, force) &&
1981                ptlrpc_nrs_req_pending_nolock(svcpt, false);
1982 }
1983
1984 /**
1985  * Returns true if there are requests available in incoming
1986  * request queue for processing and it is allowed to fetch them.
1987  * User can call it w/o any lock but need to hold ptlrpc_service::scp_req_lock
1988  * to get reliable result
1989  * \see ptlrpc_server_allow_normal
1990  * \see ptlrpc_server_allow high
1991  */
1992 static inline
1993 bool ptlrpc_server_request_pending(struct ptlrpc_service_part *svcpt,
1994                                    bool force)
1995 {
1996         return ptlrpc_server_high_pending(svcpt, force) ||
1997                ptlrpc_server_normal_pending(svcpt, force);
1998 }
1999
2000 /**
2001  * Fetch a request for processing from queue of unprocessed requests.
2002  * Favors high-priority requests.
2003  * Returns a pointer to fetched request.
2004  */
2005 static struct ptlrpc_request *
2006 ptlrpc_server_request_get(struct ptlrpc_service_part *svcpt, bool force)
2007 {
2008         struct ptlrpc_request *req = NULL;
2009
2010         ENTRY;
2011
2012         spin_lock(&svcpt->scp_req_lock);
2013
2014         if (ptlrpc_server_high_pending(svcpt, force)) {
2015                 req = ptlrpc_nrs_req_get_nolock(svcpt, true, force);
2016                 if (req != NULL) {
2017                         svcpt->scp_hreq_count++;
2018                         goto got_request;
2019                 }
2020         }
2021
2022         if (ptlrpc_server_normal_pending(svcpt, force)) {
2023                 req = ptlrpc_nrs_req_get_nolock(svcpt, false, force);
2024                 if (req != NULL) {
2025                         svcpt->scp_hreq_count = 0;
2026                         goto got_request;
2027                 }
2028         }
2029
2030         spin_unlock(&svcpt->scp_req_lock);
2031         RETURN(NULL);
2032
2033 got_request:
2034         svcpt->scp_nreqs_active++;
2035         if (req->rq_hp)
2036                 svcpt->scp_nhreqs_active++;
2037
2038         spin_unlock(&svcpt->scp_req_lock);
2039
2040         if (likely(req->rq_export))
2041                 class_export_rpc_inc(req->rq_export);
2042
2043         RETURN(req);
2044 }
2045
2046 /**
2047  * Handle freshly incoming reqs, add to timed early reply list,
2048  * pass on to regular request queue.
2049  * All incoming requests pass through here before getting into
2050  * ptlrpc_server_handle_req later on.
2051  */
2052 static int ptlrpc_server_handle_req_in(struct ptlrpc_service_part *svcpt,
2053                                        struct ptlrpc_thread *thread)
2054 {
2055         struct ptlrpc_service *svc = svcpt->scp_service;
2056         struct ptlrpc_request *req;
2057         __u32 deadline;
2058         int rc;
2059
2060         ENTRY;
2061
2062         spin_lock(&svcpt->scp_lock);
2063         if (list_empty(&svcpt->scp_req_incoming)) {
2064                 spin_unlock(&svcpt->scp_lock);
2065                 RETURN(0);
2066         }
2067
2068         req = list_entry(svcpt->scp_req_incoming.next,
2069                              struct ptlrpc_request, rq_list);
2070         list_del_init(&req->rq_list);
2071         svcpt->scp_nreqs_incoming--;
2072         /*
2073          * Consider this still a "queued" request as far as stats are
2074          * concerned
2075          */
2076         spin_unlock(&svcpt->scp_lock);
2077
2078         /* go through security check/transform */
2079         rc = sptlrpc_svc_unwrap_request(req);
2080         switch (rc) {
2081         case SECSVC_OK:
2082                 break;
2083         case SECSVC_COMPLETE:
2084                 target_send_reply(req, 0, OBD_FAIL_MDS_ALL_REPLY_NET);
2085                 goto err_req;
2086         case SECSVC_DROP:
2087                 goto err_req;
2088         default:
2089                 LBUG();
2090         }
2091
2092         /*
2093          * for null-flavored rpc, msg has been unpacked by sptlrpc, although
2094          * redo it wouldn't be harmful.
2095          */
2096         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL) {
2097                 rc = ptlrpc_unpack_req_msg(req, req->rq_reqlen);
2098                 if (rc != 0) {
2099                         CERROR("error unpacking request: ptl %d from %s x%llu\n",
2100                                svc->srv_req_portal, libcfs_id2str(req->rq_peer),
2101                                req->rq_xid);
2102                         goto err_req;
2103                 }
2104         }
2105
2106         rc = lustre_unpack_req_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
2107         if (rc) {
2108                 CERROR("error unpacking ptlrpc body: ptl %d from %s x %llu\n",
2109                        svc->srv_req_portal, libcfs_id2str(req->rq_peer),
2110                        req->rq_xid);
2111                 goto err_req;
2112         }
2113
2114         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_REQ_OPC) &&
2115             lustre_msg_get_opc(req->rq_reqmsg) == cfs_fail_val) {
2116                 CERROR("drop incoming rpc opc %u, x%llu\n",
2117                        cfs_fail_val, req->rq_xid);
2118                 goto err_req;
2119         }
2120
2121         rc = -EINVAL;
2122         if (lustre_msg_get_type(req->rq_reqmsg) != PTL_RPC_MSG_REQUEST) {
2123                 CERROR("wrong packet type received (type=%u) from %s\n",
2124                        lustre_msg_get_type(req->rq_reqmsg),
2125                        libcfs_id2str(req->rq_peer));
2126                 goto err_req;
2127         }
2128
2129         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2130         case MDS_WRITEPAGE:
2131         case OST_WRITE:
2132         case OUT_UPDATE:
2133                 req->rq_bulk_write = 1;
2134                 break;
2135         case MDS_READPAGE:
2136         case OST_READ:
2137         case MGS_CONFIG_READ:
2138                 req->rq_bulk_read = 1;
2139                 break;
2140         }
2141
2142         CDEBUG(D_RPCTRACE, "got req x%llu\n", req->rq_xid);
2143
2144         req->rq_export = class_conn2export(
2145                 lustre_msg_get_handle(req->rq_reqmsg));
2146         if (req->rq_export) {
2147                 rc = ptlrpc_check_req(req);
2148                 if (rc == 0) {
2149                         rc = sptlrpc_target_export_check(req->rq_export, req);
2150                         if (rc)
2151                                 DEBUG_REQ(D_ERROR, req,
2152                                           "DROPPING req with illegal security flavor");
2153                 }
2154
2155                 if (rc)
2156                         goto err_req;
2157                 ptlrpc_update_export_timer(req->rq_export, 0);
2158         }
2159
2160         /* req_in handling should/must be fast */
2161         if (ktime_get_real_seconds() - req->rq_arrival_time.tv_sec > 5)
2162                 DEBUG_REQ(D_WARNING, req, "Slow req_in handling %llds",
2163                           ktime_get_real_seconds() -
2164                           req->rq_arrival_time.tv_sec);
2165
2166         /* Set rpc server deadline and add it to the timed list */
2167         deadline = (lustre_msghdr_get_flags(req->rq_reqmsg) &
2168                     MSGHDR_AT_SUPPORT) ?
2169                     /* The max time the client expects us to take */
2170                     lustre_msg_get_timeout(req->rq_reqmsg) : obd_timeout;
2171
2172         req->rq_deadline = req->rq_arrival_time.tv_sec + deadline;
2173         if (unlikely(deadline == 0)) {
2174                 DEBUG_REQ(D_ERROR, req, "Dropping request with 0 timeout");
2175                 goto err_req;
2176         }
2177
2178         /* Skip early reply */
2179         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_RESEND))
2180                 req->rq_deadline += obd_timeout;
2181
2182         req->rq_svc_thread = thread;
2183         if (thread != NULL) {
2184                 /*
2185                  * initialize request session, it is needed for request
2186                  * processing by target
2187                  */
2188                 rc = lu_context_init(&req->rq_session, LCT_SERVER_SESSION |
2189                                                        LCT_NOREF);
2190                 if (rc) {
2191                         CERROR("%s: failure to initialize session: rc = %d\n",
2192                                thread->t_name, rc);
2193                         goto err_req;
2194                 }
2195                 req->rq_session.lc_thread = thread;
2196                 lu_context_enter(&req->rq_session);
2197                 thread->t_env->le_ses = &req->rq_session;
2198         }
2199
2200         ptlrpc_at_add_timed(req);
2201
2202         /* Move it over to the request processing queue */
2203         rc = ptlrpc_server_request_add(svcpt, req);
2204         if (rc)
2205                 GOTO(err_req, rc);
2206
2207         wake_up(&svcpt->scp_waitq);
2208         RETURN(1);
2209
2210 err_req:
2211         ptlrpc_server_finish_request(svcpt, req);
2212
2213         RETURN(1);
2214 }
2215
2216 /**
2217  * Main incoming request handling logic.
2218  * Calls handler function from service to do actual processing.
2219  */
2220 static int ptlrpc_server_handle_request(struct ptlrpc_service_part *svcpt,
2221                                         struct ptlrpc_thread *thread)
2222 {
2223         struct ptlrpc_service *svc = svcpt->scp_service;
2224         struct ptlrpc_request *request;
2225         ktime_t work_start;
2226         ktime_t work_end;
2227         ktime_t arrived;
2228         s64 timediff_usecs;
2229         s64 arrived_usecs;
2230         int fail_opc = 0;
2231
2232         ENTRY;
2233
2234         request = ptlrpc_server_request_get(svcpt, false);
2235         if (request == NULL)
2236                 RETURN(0);
2237
2238         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT))
2239                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT;
2240         else if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
2241                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_TIMEOUT;
2242
2243         if (unlikely(fail_opc)) {
2244                 if (request->rq_export && request->rq_ops)
2245                         OBD_FAIL_TIMEOUT(fail_opc, 4);
2246         }
2247
2248         ptlrpc_rqphase_move(request, RQ_PHASE_INTERPRET);
2249
2250         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DUMP_LOG))
2251                 libcfs_debug_dumplog();
2252
2253         work_start = ktime_get_real();
2254         arrived = timespec64_to_ktime(request->rq_arrival_time);
2255         timediff_usecs = ktime_us_delta(work_start, arrived);
2256         if (likely(svc->srv_stats != NULL)) {
2257                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
2258                                     timediff_usecs);
2259                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
2260                                     svcpt->scp_nreqs_incoming);
2261                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
2262                                     svcpt->scp_nreqs_active);
2263                 lprocfs_counter_add(svc->srv_stats, PTLRPC_TIMEOUT,
2264                                     at_get(&svcpt->scp_at_estimate));
2265         }
2266
2267         if (likely(request->rq_export)) {
2268                 if (unlikely(ptlrpc_check_req(request)))
2269                         goto put_conn;
2270                 ptlrpc_update_export_timer(request->rq_export,
2271                                            div_u64(timediff_usecs,
2272                                                    USEC_PER_SEC / 2));
2273         }
2274
2275         /*
2276          * Discard requests queued for longer than the deadline.
2277          * The deadline is increased if we send an early reply.
2278          */
2279         if (ktime_get_real_seconds() > request->rq_deadline) {
2280                 DEBUG_REQ(D_ERROR, request,
2281                           "Dropping timed-out request from %s: deadline %lld/%llds ago",
2282                           libcfs_id2str(request->rq_peer),
2283                           request->rq_deadline -
2284                           request->rq_arrival_time.tv_sec,
2285                           ktime_get_real_seconds() - request->rq_deadline);
2286                 goto put_conn;
2287         }
2288
2289         CDEBUG(D_RPCTRACE,
2290                "Handling RPC req@%p pname:cluuid+ref:pid:xid:nid:opc:job %s:%s+%d:%d:x%llu:%s:%d:%s\n",
2291                request, current->comm,
2292                (request->rq_export ?
2293                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2294                (request->rq_export ?
2295                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2296                lustre_msg_get_status(request->rq_reqmsg), request->rq_xid,
2297                libcfs_id2str(request->rq_peer),
2298                lustre_msg_get_opc(request->rq_reqmsg),
2299                lustre_msg_get_jobid(request->rq_reqmsg) ?: "");
2300
2301         if (lustre_msg_get_opc(request->rq_reqmsg) != OBD_PING)
2302                 CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_PAUSE_REQ, cfs_fail_val);
2303
2304         CDEBUG(D_NET, "got req %llu\n", request->rq_xid);
2305
2306         /* re-assign request and sesson thread to the current one */
2307         request->rq_svc_thread = thread;
2308         if (thread != NULL) {
2309                 LASSERT(request->rq_session.lc_thread == NULL);
2310                 request->rq_session.lc_thread = thread;
2311                 thread->t_env->le_ses = &request->rq_session;
2312         }
2313         svc->srv_ops.so_req_handler(request);
2314
2315         ptlrpc_rqphase_move(request, RQ_PHASE_COMPLETE);
2316
2317 put_conn:
2318         if (unlikely(ktime_get_real_seconds() > request->rq_deadline)) {
2319                 DEBUG_REQ(D_WARNING, request,
2320                           "Request took longer than estimated (%lld/%llds); client may timeout",
2321                           request->rq_deadline -
2322                           request->rq_arrival_time.tv_sec,
2323                           ktime_get_real_seconds() - request->rq_deadline);
2324         }
2325
2326         work_end = ktime_get_real();
2327         timediff_usecs = ktime_us_delta(work_end, work_start);
2328         arrived_usecs = ktime_us_delta(work_end, arrived);
2329         CDEBUG(D_RPCTRACE,
2330                "Handled RPC req@%p pname:cluuid+ref:pid:xid:nid:opc:job %s:%s+%d:%d:x%llu:%s:%d:%s Request processed in %lldus (%lldus total) trans %llu rc %d/%d\n",
2331                request, current->comm,
2332                (request->rq_export ?
2333                (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2334                (request->rq_export ?
2335                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2336                lustre_msg_get_status(request->rq_reqmsg),
2337                request->rq_xid,
2338                libcfs_id2str(request->rq_peer),
2339                lustre_msg_get_opc(request->rq_reqmsg),
2340                lustre_msg_get_jobid(request->rq_reqmsg) ?: "",
2341                timediff_usecs,
2342                arrived_usecs,
2343                (request->rq_repmsg ?
2344                lustre_msg_get_transno(request->rq_repmsg) :
2345                request->rq_transno),
2346                request->rq_status,
2347                (request->rq_repmsg ?
2348                lustre_msg_get_status(request->rq_repmsg) : -999));
2349         if (likely(svc->srv_stats != NULL && request->rq_reqmsg != NULL)) {
2350                 __u32 op = lustre_msg_get_opc(request->rq_reqmsg);
2351                 int opc = opcode_offset(op);
2352
2353                 if (opc > 0 && !(op == LDLM_ENQUEUE || op == MDS_REINT)) {
2354                         LASSERT(opc < LUSTRE_MAX_OPCODES);
2355                         lprocfs_counter_add(svc->srv_stats,
2356                                             opc + EXTRA_MAX_OPCODES,
2357                                             timediff_usecs);
2358                 }
2359         }
2360         if (unlikely(request->rq_early_count)) {
2361                 DEBUG_REQ(D_ADAPTTO, request,
2362                           "sent %d early replies before finishing in %llds",
2363                           request->rq_early_count,
2364                           div_u64(arrived_usecs, USEC_PER_SEC));
2365         }
2366
2367         ptlrpc_server_finish_active_request(svcpt, request);
2368
2369         RETURN(1);
2370 }
2371
2372 /**
2373  * An internal function to process a single reply state object.
2374  */
2375 static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
2376 {
2377         struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
2378         struct ptlrpc_service *svc = svcpt->scp_service;
2379         struct obd_export *exp;
2380         int nlocks;
2381         int been_handled;
2382
2383         ENTRY;
2384
2385         exp = rs->rs_export;
2386
2387         LASSERT(rs->rs_difficult);
2388         LASSERT(rs->rs_scheduled);
2389         LASSERT(list_empty(&rs->rs_list));
2390
2391         /*
2392          * The disk commit callback holds exp_uncommitted_replies_lock while it
2393          * iterates over newly committed replies, removing them from
2394          * exp_uncommitted_replies.  It then drops this lock and schedules the
2395          * replies it found for handling here.
2396          *
2397          * We can avoid contention for exp_uncommitted_replies_lock between the
2398          * HRT threads and further commit callbacks by checking rs_committed
2399          * which is set in the commit callback while it holds both
2400          * rs_lock and exp_uncommitted_reples.
2401          *
2402          * If we see rs_committed clear, the commit callback _may_ not have
2403          * handled this reply yet and we race with it to grab
2404          * exp_uncommitted_replies_lock before removing the reply from
2405          * exp_uncommitted_replies.  Note that if we lose the race and the
2406          * reply has already been removed, list_del_init() is a noop.
2407          *
2408          * If we see rs_committed set, we know the commit callback is handling,
2409          * or has handled this reply since store reordering might allow us to
2410          * see rs_committed set out of sequence.  But since this is done
2411          * holding rs_lock, we can be sure it has all completed once we hold
2412          * rs_lock, which we do right next.
2413          */
2414         if (!rs->rs_committed) {
2415                 /*
2416                  * if rs was commited, no need to convert locks, don't check
2417                  * rs_committed here because rs may never be added into
2418                  * exp_uncommitted_replies and this flag never be set, see
2419                  * target_send_reply()
2420                  */
2421                 if (rs->rs_convert_lock &&
2422                     rs->rs_transno > exp->exp_last_committed) {
2423                         struct ldlm_lock *lock;
2424                         struct ldlm_lock *ack_locks[RS_MAX_LOCKS] = { NULL };
2425
2426                         spin_lock(&rs->rs_lock);
2427                         if (rs->rs_convert_lock &&
2428                             rs->rs_transno > exp->exp_last_committed) {
2429                                 nlocks = rs->rs_nlocks;
2430                                 while (nlocks-- > 0) {
2431                                         /*
2432                                          * NB don't assume rs is always handled
2433                                          * by the same service thread (see
2434                                          * ptlrpc_hr_select, so REP-ACK hr may
2435                                          * race with trans commit, while the
2436                                          * latter will release locks, get locks
2437                                          * here early to convert to COS mode
2438                                          * safely.
2439                                          */
2440                                         lock = ldlm_handle2lock(
2441                                                         &rs->rs_locks[nlocks]);
2442                                         LASSERT(lock);
2443                                         ack_locks[nlocks] = lock;
2444                                         rs->rs_modes[nlocks] = LCK_COS;
2445                                 }
2446                                 nlocks = rs->rs_nlocks;
2447                                 rs->rs_convert_lock = 0;
2448                                 /*
2449                                  * clear rs_scheduled so that commit callback
2450                                  * can schedule again
2451                                  */
2452                                 rs->rs_scheduled = 0;
2453                                 spin_unlock(&rs->rs_lock);
2454
2455                                 while (nlocks-- > 0) {
2456                                         lock = ack_locks[nlocks];
2457                                         ldlm_lock_mode_downgrade(lock, LCK_COS);
2458                                         LDLM_LOCK_PUT(lock);
2459                                 }
2460                                 RETURN(0);
2461                         }
2462                         spin_unlock(&rs->rs_lock);
2463                 }
2464
2465                 spin_lock(&exp->exp_uncommitted_replies_lock);
2466                 list_del_init(&rs->rs_obd_list);
2467                 spin_unlock(&exp->exp_uncommitted_replies_lock);
2468         }
2469
2470         spin_lock(&exp->exp_lock);
2471         /* Noop if removed already */
2472         list_del_init(&rs->rs_exp_list);
2473         spin_unlock(&exp->exp_lock);
2474
2475         spin_lock(&rs->rs_lock);
2476
2477         been_handled = rs->rs_handled;
2478         rs->rs_handled = 1;
2479
2480         nlocks = rs->rs_nlocks; /* atomic "steal", but */
2481         rs->rs_nlocks = 0; /* locks still on rs_locks! */
2482
2483         if (nlocks == 0 && !been_handled) {
2484                 /*
2485                  * If we see this, we should already have seen the warning
2486                  * in mds_steal_ack_locks()
2487                  */
2488                 CDEBUG(D_HA,
2489                        "All locks stolen from rs %p x%lld.t%lld o%d NID %s\n",
2490                        rs, rs->rs_xid, rs->rs_transno, rs->rs_opc,
2491                        libcfs_nid2str(exp->exp_connection->c_peer.nid));
2492         }
2493
2494         if ((!been_handled && rs->rs_on_net) || nlocks > 0) {
2495                 spin_unlock(&rs->rs_lock);
2496
2497                 if (!been_handled && rs->rs_on_net) {
2498                         LNetMDUnlink(rs->rs_md_h);
2499                         /* Ignore return code; we're racing with completion */
2500                 }
2501
2502                 while (nlocks-- > 0)
2503                         ldlm_lock_decref(&rs->rs_locks[nlocks],
2504                                          rs->rs_modes[nlocks]);
2505
2506                 spin_lock(&rs->rs_lock);
2507         }
2508
2509         rs->rs_scheduled = 0;
2510         rs->rs_convert_lock = 0;
2511
2512         if (!rs->rs_on_net) {
2513                 /* Off the net */
2514                 spin_unlock(&rs->rs_lock);
2515
2516                 class_export_put(exp);
2517                 rs->rs_export = NULL;
2518                 ptlrpc_rs_decref(rs);
2519                 if (atomic_dec_and_test(&svcpt->scp_nreps_difficult) &&
2520                     svc->srv_is_stopping)
2521                         wake_up_all(&svcpt->scp_waitq);
2522                 RETURN(1);
2523         }
2524
2525         /* still on the net; callback will schedule */
2526         spin_unlock(&rs->rs_lock);
2527         RETURN(1);
2528 }
2529
2530
2531 static void ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
2532 {
2533         int avail = svcpt->scp_nrqbds_posted;
2534         int low_water = test_req_buffer_pressure ? 0 :
2535                         svcpt->scp_service->srv_nbuf_per_group / 2;
2536
2537         /* NB I'm not locking; just looking. */
2538
2539         /*
2540          * CAVEAT EMPTOR: We might be allocating buffers here because we've
2541          * allowed the request history to grow out of control.  We could put a
2542          * sanity check on that here and cull some history if we need the
2543          * space.
2544          */
2545
2546         if (avail <= low_water)
2547                 ptlrpc_grow_req_bufs(svcpt, 1);
2548
2549         if (svcpt->scp_service->srv_stats) {
2550                 lprocfs_counter_add(svcpt->scp_service->srv_stats,
2551                                     PTLRPC_REQBUF_AVAIL_CNTR, avail);
2552         }
2553 }
2554
2555 static inline int ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
2556 {
2557         return svcpt->scp_nreqs_active <
2558                svcpt->scp_nthrs_running - 1 -
2559                (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL);
2560 }
2561
2562 /**
2563  * allowed to create more threads
2564  * user can call it w/o any lock but need to hold
2565  * ptlrpc_service_part::scp_lock to get reliable result
2566  */
2567 static inline int ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
2568 {
2569         return svcpt->scp_nthrs_running +
2570                svcpt->scp_nthrs_starting <
2571                svcpt->scp_service->srv_nthrs_cpt_limit;
2572 }
2573
2574 /**
2575  * too many requests and allowed to create more threads
2576  */
2577 static inline int ptlrpc_threads_need_create(struct ptlrpc_service_part *svcpt)
2578 {
2579         return !ptlrpc_threads_enough(svcpt) &&
2580                 ptlrpc_threads_increasable(svcpt);
2581 }
2582
2583 static inline int ptlrpc_thread_stopping(struct ptlrpc_thread *thread)
2584 {
2585         return thread_is_stopping(thread) ||
2586                thread->t_svcpt->scp_service->srv_is_stopping;
2587 }
2588
2589 /* stop the highest numbered thread if there are too many threads running */
2590 static inline bool ptlrpc_thread_should_stop(struct ptlrpc_thread *thread)
2591 {
2592         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2593
2594         return thread->t_id >= svcpt->scp_service->srv_nthrs_cpt_limit &&
2595                 thread->t_id == svcpt->scp_thr_nextid - 1;
2596 }
2597
2598 static void ptlrpc_stop_thread(struct ptlrpc_thread *thread)
2599 {
2600         CDEBUG(D_INFO, "Stopping thread %s #%u\n",
2601                thread->t_svcpt->scp_service->srv_thread_name, thread->t_id);
2602         thread_add_flags(thread, SVC_STOPPING);
2603 }
2604
2605 static inline void ptlrpc_thread_stop(struct ptlrpc_thread *thread)
2606 {
2607         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2608
2609         spin_lock(&svcpt->scp_lock);
2610         if (ptlrpc_thread_should_stop(thread)) {
2611                 ptlrpc_stop_thread(thread);
2612                 svcpt->scp_thr_nextid--;
2613         }
2614         spin_unlock(&svcpt->scp_lock);
2615 }
2616
2617 static inline int ptlrpc_rqbd_pending(struct ptlrpc_service_part *svcpt)
2618 {
2619         return !list_empty(&svcpt->scp_rqbd_idle) &&
2620                svcpt->scp_rqbd_timeout == 0;
2621 }
2622
2623 static inline int
2624 ptlrpc_at_check(struct ptlrpc_service_part *svcpt)
2625 {
2626         return svcpt->scp_at_check;
2627 }
2628
2629 /*
2630  * If a thread runs too long or spends to much time on a single request,
2631  * we want to know about it, so we set up a delayed work item as a watchdog.
2632  * If it fires, we display a stack trace of the delayed thread,
2633  * providing we aren't rate-limited
2634  *
2635  * Watchdog stack traces are limited to 3 per 'libcfs_watchdog_ratelimit'
2636  * seconds
2637  */
2638 static struct ratelimit_state watchdog_limit;
2639
2640 static void ptlrpc_watchdog_fire(struct work_struct *w)
2641 {
2642         struct ptlrpc_thread *thread = container_of(w, struct ptlrpc_thread,
2643                                                     t_watchdog.work);
2644         u64 ms_lapse = ktime_ms_delta(ktime_get(), thread->t_touched);
2645         u32 ms_frac = do_div(ms_lapse, MSEC_PER_SEC);
2646
2647         /* ___ratelimit() returns true if the action is NOT ratelimited */
2648         if (__ratelimit(&watchdog_limit)) {
2649                 /* below message is checked in sanity-quota.sh test_6,18 */
2650                 LCONSOLE_WARN("%s: service thread pid %u was inactive for %llu.%03u seconds. The thread might be hung, or it might only be slow and will resume later. Dumping the stack trace for debugging purposes:\n",
2651                               thread->t_task->comm, thread->t_task->pid,
2652                               ms_lapse, ms_frac);
2653
2654                 libcfs_debug_dumpstack(thread->t_task);
2655         } else {
2656                 /* below message is checked in sanity-quota.sh test_6,18 */
2657                 LCONSOLE_WARN("%s: service thread pid %u was inactive for %llu.%03u seconds. Watchdog stack traces are limited to 3 per %u seconds, skipping this one.\n",
2658                               thread->t_task->comm, thread->t_task->pid,
2659                               ms_lapse, ms_frac, libcfs_watchdog_ratelimit);
2660         }
2661 }
2662
2663 static void ptlrpc_watchdog_init(struct delayed_work *work, timeout_t timeout)
2664 {
2665         INIT_DELAYED_WORK(work, ptlrpc_watchdog_fire);
2666         schedule_delayed_work(work, cfs_time_seconds(timeout));
2667 }
2668
2669 static void ptlrpc_watchdog_disable(struct delayed_work *work)
2670 {
2671         cancel_delayed_work_sync(work);
2672 }
2673
2674 static void ptlrpc_watchdog_touch(struct delayed_work *work, timeout_t timeout)
2675 {
2676         struct ptlrpc_thread *thread = container_of(&work->work,
2677                                                     struct ptlrpc_thread,
2678                                                     t_watchdog.work);
2679         thread->t_touched = ktime_get();
2680         mod_delayed_work(system_wq, work, cfs_time_seconds(timeout));
2681 }
2682
2683 /**
2684  * requests wait on preprocessing
2685  * user can call it w/o any lock but need to hold
2686  * ptlrpc_service_part::scp_lock to get reliable result
2687  */
2688 static inline int
2689 ptlrpc_server_request_incoming(struct ptlrpc_service_part *svcpt)
2690 {
2691         return !list_empty(&svcpt->scp_req_incoming);
2692 }
2693
2694 static __attribute__((__noinline__)) int
2695 ptlrpc_wait_event(struct ptlrpc_service_part *svcpt,
2696                   struct ptlrpc_thread *thread)
2697 {
2698         ptlrpc_watchdog_disable(&thread->t_watchdog);
2699
2700         cond_resched();
2701
2702         if (svcpt->scp_rqbd_timeout == 0)
2703                 /* Don't exit while there are replies to be handled */
2704                 wait_event_idle_exclusive_lifo(
2705                         svcpt->scp_waitq,
2706                         ptlrpc_thread_stopping(thread) ||
2707                         ptlrpc_server_request_incoming(svcpt) ||
2708                         ptlrpc_server_request_pending(svcpt, false) ||
2709                         ptlrpc_rqbd_pending(svcpt) ||
2710                         ptlrpc_at_check(svcpt));
2711         else if (wait_event_idle_exclusive_lifo_timeout(
2712                          svcpt->scp_waitq,
2713                          ptlrpc_thread_stopping(thread) ||
2714                          ptlrpc_server_request_incoming(svcpt) ||
2715                          ptlrpc_server_request_pending(svcpt, false) ||
2716                          ptlrpc_rqbd_pending(svcpt) ||
2717                          ptlrpc_at_check(svcpt),
2718                          svcpt->scp_rqbd_timeout) == 0)
2719                 svcpt->scp_rqbd_timeout = 0;
2720
2721         if (ptlrpc_thread_stopping(thread))
2722                 return -EINTR;
2723
2724         ptlrpc_watchdog_touch(&thread->t_watchdog,
2725                               ptlrpc_server_get_timeout(svcpt));
2726         return 0;
2727 }
2728
2729 /**
2730  * Main thread body for service threads.
2731  * Waits in a loop waiting for new requests to process to appear.
2732  * Every time an incoming requests is added to its queue, a waitq
2733  * is woken up and one of the threads will handle it.
2734  */
2735 static int ptlrpc_main(void *arg)
2736 {
2737         struct ptlrpc_thread *thread = (struct ptlrpc_thread *)arg;
2738         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2739         struct ptlrpc_service *svc = svcpt->scp_service;
2740         struct ptlrpc_reply_state *rs;
2741         struct group_info *ginfo = NULL;
2742         struct lu_env *env;
2743         int counter = 0, rc = 0;
2744
2745         ENTRY;
2746
2747         thread->t_task = current;
2748         thread->t_pid = current->pid;
2749         unshare_fs_struct();
2750
2751         if (svc->srv_cpt_bind) {
2752                 rc = cfs_cpt_bind(svc->srv_cptable, svcpt->scp_cpt);
2753                 if (rc != 0) {
2754                         CWARN("%s: failed to bind %s on CPT %d\n",
2755                               svc->srv_name, thread->t_name, svcpt->scp_cpt);
2756                 }
2757         }
2758
2759         ginfo = groups_alloc(0);
2760         if (!ginfo)
2761                 GOTO(out, rc = -ENOMEM);
2762
2763         set_current_groups(ginfo);
2764         put_group_info(ginfo);
2765
2766         if (svc->srv_ops.so_thr_init != NULL) {
2767                 rc = svc->srv_ops.so_thr_init(thread);
2768                 if (rc)
2769                         GOTO(out, rc);
2770         }
2771
2772         OBD_ALLOC_PTR(env);
2773         if (env == NULL)
2774                 GOTO(out_srv_fini, rc = -ENOMEM);
2775         rc = lu_env_add(env);
2776         if (rc)
2777                 GOTO(out_env, rc);
2778
2779         rc = lu_context_init(&env->le_ctx,
2780                              svc->srv_ctx_tags|LCT_REMEMBER|LCT_NOREF);
2781         if (rc)
2782                 GOTO(out_env_remove, rc);
2783
2784         thread->t_env = env;
2785         env->le_ctx.lc_thread = thread;
2786         env->le_ctx.lc_cookie = 0x6;
2787
2788         while (!list_empty(&svcpt->scp_rqbd_idle)) {
2789                 rc = ptlrpc_server_post_idle_rqbds(svcpt);
2790                 if (rc >= 0)
2791                         continue;
2792
2793                 CERROR("Failed to post rqbd for %s on CPT %d: %d\n",
2794                         svc->srv_name, svcpt->scp_cpt, rc);
2795                 GOTO(out_ctx_fini, rc);
2796         }
2797
2798         /* Alloc reply state structure for this one */
2799         OBD_ALLOC_LARGE(rs, svc->srv_max_reply_size);
2800         if (!rs)
2801                 GOTO(out_ctx_fini, rc = -ENOMEM);
2802
2803         spin_lock(&svcpt->scp_lock);
2804
2805         LASSERT(thread_is_starting(thread));
2806         thread_clear_flags(thread, SVC_STARTING);
2807
2808         LASSERT(svcpt->scp_nthrs_starting == 1);
2809         svcpt->scp_nthrs_starting--;
2810
2811         /*
2812          * SVC_STOPPING may already be set here if someone else is trying
2813          * to stop the service while this new thread has been dynamically
2814          * forked. We still set SVC_RUNNING to let our creator know that
2815          * we are now running, however we will exit as soon as possible
2816          */
2817         thread_add_flags(thread, SVC_RUNNING);
2818         svcpt->scp_nthrs_running++;
2819         spin_unlock(&svcpt->scp_lock);
2820
2821         /* wake up our creator in case he's still waiting. */
2822         wake_up(&thread->t_ctl_waitq);
2823
2824         thread->t_touched = ktime_get();
2825         ptlrpc_watchdog_init(&thread->t_watchdog,
2826                          ptlrpc_server_get_timeout(svcpt));
2827
2828         spin_lock(&svcpt->scp_rep_lock);
2829         list_add(&rs->rs_list, &svcpt->scp_rep_idle);
2830         wake_up(&svcpt->scp_rep_waitq);
2831         spin_unlock(&svcpt->scp_rep_lock);
2832
2833         CDEBUG(D_NET, "service thread %d (#%d) started\n", thread->t_id,
2834                svcpt->scp_nthrs_running);
2835
2836         /* XXX maintain a list of all managed devices: insert here */
2837         while (!ptlrpc_thread_stopping(thread)) {
2838                 if (ptlrpc_wait_event(svcpt, thread))
2839                         break;
2840
2841                 ptlrpc_check_rqbd_pool(svcpt);
2842
2843                 if (ptlrpc_threads_need_create(svcpt)) {
2844                         /* Ignore return code - we tried... */
2845                         ptlrpc_start_thread(svcpt, 0);
2846                 }
2847
2848                 /* reset le_ses to initial state */
2849                 env->le_ses = NULL;
2850                 /* Refill the context before execution to make sure
2851                  * all thread keys are allocated */
2852                 lu_env_refill(env);
2853                 /* Process all incoming reqs before handling any */
2854                 if (ptlrpc_server_request_incoming(svcpt)) {
2855                         lu_context_enter(&env->le_ctx);
2856                         ptlrpc_server_handle_req_in(svcpt, thread);
2857                         lu_context_exit(&env->le_ctx);
2858
2859                         /* but limit ourselves in case of flood */
2860                         if (counter++ < 100)
2861                                 continue;
2862                         counter = 0;
2863                 }
2864
2865                 if (ptlrpc_at_check(svcpt))
2866                         ptlrpc_at_check_timed(svcpt);
2867
2868                 if (ptlrpc_server_request_pending(svcpt, false)) {
2869                         lu_context_enter(&env->le_ctx);
2870                         ptlrpc_server_handle_request(svcpt, thread);
2871                         lu_context_exit(&env->le_ctx);
2872                 }
2873
2874                 if (ptlrpc_rqbd_pending(svcpt) &&
2875                     ptlrpc_server_post_idle_rqbds(svcpt) < 0) {
2876                         /*
2877                          * I just failed to repost request buffers.
2878                          * Wait for a timeout (unless something else
2879                          * happens) before I try again
2880                          */
2881                         svcpt->scp_rqbd_timeout = cfs_time_seconds(1) / 10;
2882                         CDEBUG(D_RPCTRACE, "Posted buffers: %d\n",
2883                                svcpt->scp_nrqbds_posted);
2884                 }
2885                 /*
2886                  * If the number of threads has been tuned downward and this
2887                  * thread should be stopped, then stop in reverse order so the
2888                  * the threads always have contiguous thread index values.
2889                  */
2890                 if (unlikely(ptlrpc_thread_should_stop(thread)))
2891                         ptlrpc_thread_stop(thread);
2892         }
2893
2894         ptlrpc_watchdog_disable(&thread->t_watchdog);
2895
2896 out_ctx_fini:
2897         lu_context_fini(&env->le_ctx);
2898 out_env_remove:
2899         lu_env_remove(env);
2900 out_env:
2901         OBD_FREE_PTR(env);
2902 out_srv_fini:
2903         /* deconstruct service thread state created by ptlrpc_start_thread() */
2904         if (svc->srv_ops.so_thr_done != NULL)
2905                 svc->srv_ops.so_thr_done(thread);
2906 out:
2907         CDEBUG(D_RPCTRACE, "%s: service thread [%p:%u] %d exiting: rc = %d\n",
2908                thread->t_name, thread, thread->t_pid, thread->t_id, rc);
2909         spin_lock(&svcpt->scp_lock);
2910         if (thread_test_and_clear_flags(thread, SVC_STARTING))
2911                 svcpt->scp_nthrs_starting--;
2912
2913         if (thread_test_and_clear_flags(thread, SVC_RUNNING)) {
2914                 /* must know immediately */
2915                 svcpt->scp_nthrs_running--;
2916         }
2917
2918         thread->t_id = rc;
2919         thread_add_flags(thread, SVC_STOPPED);
2920
2921         wake_up(&thread->t_ctl_waitq);
2922         spin_unlock(&svcpt->scp_lock);
2923
2924         return rc;
2925 }
2926
2927 static int hrt_dont_sleep(struct ptlrpc_hr_thread *hrt,
2928                           struct list_head *replies)
2929 {
2930         int result;
2931
2932         spin_lock(&hrt->hrt_lock);
2933
2934         list_splice_init(&hrt->hrt_queue, replies);
2935         result = ptlrpc_hr.hr_stopping || !list_empty(replies);
2936
2937         spin_unlock(&hrt->hrt_lock);
2938         return result;
2939 }
2940
2941 /**
2942  * Main body of "handle reply" function.
2943  * It processes acked reply states
2944  */
2945 static int ptlrpc_hr_main(void *arg)
2946 {
2947         struct ptlrpc_hr_thread *hrt = (struct ptlrpc_hr_thread *)arg;
2948         struct ptlrpc_hr_partition *hrp = hrt->hrt_partition;
2949         LIST_HEAD(replies);
2950         struct lu_env *env;
2951         int rc;
2952
2953         OBD_ALLOC_PTR(env);
2954         if (env == NULL)
2955                 RETURN(-ENOMEM);
2956
2957         unshare_fs_struct();
2958
2959         rc = cfs_cpt_bind(ptlrpc_hr.hr_cpt_table, hrp->hrp_cpt);
2960         if (rc != 0) {
2961                 char threadname[20];
2962
2963                 snprintf(threadname, sizeof(threadname), "ptlrpc_hr%02d_%03d",
2964                          hrp->hrp_cpt, hrt->hrt_id);
2965                 CWARN("Failed to bind %s on CPT %d of CPT table %p: rc = %d\n",
2966                       threadname, hrp->hrp_cpt, ptlrpc_hr.hr_cpt_table, rc);
2967         }
2968
2969         rc = lu_context_init(&env->le_ctx, LCT_MD_THREAD | LCT_DT_THREAD |
2970                              LCT_REMEMBER | LCT_NOREF);
2971         if (rc)
2972                 GOTO(out_env, rc);
2973
2974         rc = lu_env_add(env);
2975         if (rc)
2976                 GOTO(out_ctx_fini, rc);
2977
2978         atomic_inc(&hrp->hrp_nstarted);
2979         wake_up(&ptlrpc_hr.hr_waitq);
2980
2981         while (!ptlrpc_hr.hr_stopping) {
2982                 wait_event_idle(hrt->hrt_waitq, hrt_dont_sleep(hrt, &replies));
2983
2984                 while (!list_empty(&replies)) {
2985                         struct ptlrpc_reply_state *rs;
2986
2987                         rs = list_entry(replies.prev,
2988                                         struct ptlrpc_reply_state,
2989                                         rs_list);
2990                         list_del_init(&rs->rs_list);
2991                         /* refill keys if needed */
2992                         lu_env_refill(env);
2993                         lu_context_enter(&env->le_ctx);
2994                         ptlrpc_handle_rs(rs);
2995                         lu_context_exit(&env->le_ctx);
2996                 }
2997         }
2998
2999         atomic_inc(&hrp->hrp_nstopped);
3000         wake_up(&ptlrpc_hr.hr_waitq);
3001
3002         lu_env_remove(env);
3003 out_ctx_fini:
3004         lu_context_fini(&env->le_ctx);
3005 out_env:
3006         OBD_FREE_PTR(env);
3007         return 0;
3008 }
3009
3010 static void ptlrpc_stop_hr_threads(void)
3011 {
3012         struct ptlrpc_hr_partition *hrp;
3013         int i;
3014         int j;
3015
3016         ptlrpc_hr.hr_stopping = 1;
3017
3018         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3019                 if (hrp->hrp_thrs == NULL)
3020                         continue; /* uninitialized */
3021                 for (j = 0; j < hrp->hrp_nthrs; j++)
3022                         wake_up_all(&hrp->hrp_thrs[j].hrt_waitq);
3023         }
3024
3025         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3026                 if (hrp->hrp_thrs == NULL)
3027                         continue; /* uninitialized */
3028                 wait_event(ptlrpc_hr.hr_waitq,
3029                                atomic_read(&hrp->hrp_nstopped) ==
3030                                atomic_read(&hrp->hrp_nstarted));
3031         }
3032 }
3033
3034 static int ptlrpc_start_hr_threads(void)
3035 {
3036         struct ptlrpc_hr_partition *hrp;
3037         int i;
3038         int j;
3039
3040         ENTRY;
3041
3042         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3043                 int     rc = 0;
3044
3045                 for (j = 0; j < hrp->hrp_nthrs; j++) {
3046                         struct ptlrpc_hr_thread *hrt = &hrp->hrp_thrs[j];
3047                         struct task_struct *task;
3048
3049                         task = kthread_run(ptlrpc_hr_main,
3050                                            &hrp->hrp_thrs[j],
3051                                            "ptlrpc_hr%02d_%03d",
3052                                            hrp->hrp_cpt,
3053                                            hrt->hrt_id);
3054                         if (IS_ERR(task)) {
3055                                 rc = PTR_ERR(task);
3056                                 break;
3057                         }
3058                 }
3059
3060                 wait_event(ptlrpc_hr.hr_waitq,
3061                            atomic_read(&hrp->hrp_nstarted) == j);
3062
3063                 if (rc < 0) {
3064                         CERROR("cannot start reply handler thread %d:%d: rc = %d\n",
3065                                i, j, rc);
3066                         ptlrpc_stop_hr_threads();
3067                         RETURN(rc);
3068                 }
3069         }
3070
3071         RETURN(0);
3072 }
3073
3074 static void ptlrpc_svcpt_stop_threads(struct ptlrpc_service_part *svcpt)
3075 {
3076         struct ptlrpc_thread *thread;
3077         LIST_HEAD(zombie);
3078
3079         ENTRY;
3080
3081         CDEBUG(D_INFO, "Stopping threads for service %s\n",
3082                svcpt->scp_service->srv_name);
3083
3084         spin_lock(&svcpt->scp_lock);
3085         /* let the thread know that we would like it to stop asap */
3086         list_for_each_entry(thread, &svcpt->scp_threads, t_link)
3087                 ptlrpc_stop_thread(thread);
3088
3089         wake_up_all(&svcpt->scp_waitq);
3090
3091         while (!list_empty(&svcpt->scp_threads)) {
3092                 thread = list_entry(svcpt->scp_threads.next,
3093                                         struct ptlrpc_thread, t_link);
3094                 if (thread_is_stopped(thread)) {
3095                         list_move(&thread->t_link, &zombie);
3096                         continue;
3097                 }
3098                 spin_unlock(&svcpt->scp_lock);
3099
3100                 CDEBUG(D_INFO, "waiting for stopping-thread %s #%u\n",
3101                        svcpt->scp_service->srv_thread_name, thread->t_id);
3102                 wait_event_idle(thread->t_ctl_waitq,
3103                                 thread_is_stopped(thread));
3104
3105                 spin_lock(&svcpt->scp_lock);
3106         }
3107
3108         spin_unlock(&svcpt->scp_lock);
3109
3110         while (!list_empty(&zombie)) {
3111                 thread = list_entry(zombie.next,
3112                                         struct ptlrpc_thread, t_link);
3113                 list_del(&thread->t_link);
3114                 OBD_FREE_PTR(thread);
3115         }
3116         EXIT;
3117 }
3118
3119 /**
3120  * Stops all threads of a particular service \a svc
3121  */
3122 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
3123 {
3124         struct ptlrpc_service_part *svcpt;
3125         int i;
3126
3127         ENTRY;
3128
3129         ptlrpc_service_for_each_part(svcpt, i, svc) {
3130                 if (svcpt->scp_service != NULL)
3131                         ptlrpc_svcpt_stop_threads(svcpt);
3132         }
3133
3134         EXIT;
3135 }
3136
3137 int ptlrpc_start_threads(struct ptlrpc_service *svc)
3138 {
3139         int rc = 0;
3140         int i;
3141         int j;
3142
3143         ENTRY;
3144
3145         /* We require 2 threads min, see note in ptlrpc_server_handle_request */
3146         LASSERT(svc->srv_nthrs_cpt_init >= PTLRPC_NTHRS_INIT);
3147
3148         for (i = 0; i < svc->srv_ncpts; i++) {
3149                 for (j = 0; j < svc->srv_nthrs_cpt_init; j++) {
3150                         rc = ptlrpc_start_thread(svc->srv_parts[i], 1);
3151                         if (rc == 0)
3152                                 continue;
3153
3154                         if (rc != -EMFILE)
3155                                 goto failed;
3156                         /* We have enough threads, don't start more. b=15759 */
3157                         break;
3158                 }
3159         }
3160
3161         RETURN(0);
3162  failed:
3163         CERROR("cannot start %s thread #%d_%d: rc %d\n",
3164                svc->srv_thread_name, i, j, rc);
3165         ptlrpc_stop_all_threads(svc);
3166         RETURN(rc);
3167 }
3168
3169 int ptlrpc_start_thread(struct ptlrpc_service_part *svcpt, int wait)
3170 {
3171         struct ptlrpc_thread *thread;
3172         struct ptlrpc_service *svc;
3173         struct task_struct *task;
3174         int rc;
3175
3176         ENTRY;
3177
3178         LASSERT(svcpt != NULL);
3179
3180         svc = svcpt->scp_service;
3181
3182         CDEBUG(D_RPCTRACE, "%s[%d] started %d min %d max %d\n",
3183                svc->srv_name, svcpt->scp_cpt, svcpt->scp_nthrs_running,
3184                svc->srv_nthrs_cpt_init, svc->srv_nthrs_cpt_limit);
3185
3186  again:
3187         if (unlikely(svc->srv_is_stopping))
3188                 RETURN(-ESRCH);
3189
3190         if (!ptlrpc_threads_increasable(svcpt) ||
3191             (OBD_FAIL_CHECK(OBD_FAIL_TGT_TOOMANY_THREADS) &&
3192              svcpt->scp_nthrs_running == svc->srv_nthrs_cpt_init - 1))
3193                 RETURN(-EMFILE);
3194
3195         OBD_CPT_ALLOC_PTR(thread, svc->srv_cptable, svcpt->scp_cpt);
3196         if (thread == NULL)
3197                 RETURN(-ENOMEM);
3198         init_waitqueue_head(&thread->t_ctl_waitq);
3199
3200         spin_lock(&svcpt->scp_lock);
3201         if (!ptlrpc_threads_increasable(svcpt)) {
3202                 spin_unlock(&svcpt->scp_lock);
3203                 OBD_FREE_PTR(thread);
3204                 RETURN(-EMFILE);
3205         }
3206
3207         if (svcpt->scp_nthrs_starting != 0) {
3208                 /*
3209                  * serialize starting because some modules (obdfilter)
3210                  * might require unique and contiguous t_id
3211                  */
3212                 LASSERT(svcpt->scp_nthrs_starting == 1);
3213                 spin_unlock(&svcpt->scp_lock);
3214                 OBD_FREE_PTR(thread);
3215                 if (wait) {
3216                         CDEBUG(D_INFO, "Waiting for creating thread %s #%d\n",
3217                                svc->srv_thread_name, svcpt->scp_thr_nextid);
3218                         schedule();
3219                         goto again;
3220                 }
3221
3222                 CDEBUG(D_INFO, "Creating thread %s #%d race, retry later\n",
3223                        svc->srv_thread_name, svcpt->scp_thr_nextid);
3224                 RETURN(-EAGAIN);
3225         }
3226
3227         svcpt->scp_nthrs_starting++;
3228         thread->t_id = svcpt->scp_thr_nextid++;
3229         thread_add_flags(thread, SVC_STARTING);
3230         thread->t_svcpt = svcpt;
3231
3232         list_add(&thread->t_link, &svcpt->scp_threads);
3233         spin_unlock(&svcpt->scp_lock);
3234
3235         if (svcpt->scp_cpt >= 0) {
3236                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s%02d_%03d",
3237                          svc->srv_thread_name, svcpt->scp_cpt, thread->t_id);
3238         } else {
3239                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s_%04d",
3240                          svc->srv_thread_name, thread->t_id);
3241         }
3242
3243         CDEBUG(D_RPCTRACE, "starting thread '%s'\n", thread->t_name);
3244         task = kthread_run(ptlrpc_main, thread, "%s", thread->t_name);
3245         if (IS_ERR(task)) {
3246                 rc = PTR_ERR(task);
3247                 CERROR("cannot start thread '%s': rc = %d\n",
3248                        thread->t_name, rc);
3249                 spin_lock(&svcpt->scp_lock);
3250                 --svcpt->scp_nthrs_starting;
3251                 if (thread_is_stopping(thread)) {
3252                         /*
3253                          * this ptlrpc_thread is being hanled
3254                          * by ptlrpc_svcpt_stop_threads now
3255                          */
3256                         thread_add_flags(thread, SVC_STOPPED);
3257                         wake_up(&thread->t_ctl_waitq);
3258                         spin_unlock(&svcpt->scp_lock);
3259                 } else {
3260                         list_del(&thread->t_link);
3261                         spin_unlock(&svcpt->scp_lock);
3262                         OBD_FREE_PTR(thread);
3263                 }
3264                 RETURN(rc);
3265         }
3266
3267         if (!wait)
3268                 RETURN(0);
3269
3270         wait_event_idle(thread->t_ctl_waitq,