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