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