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