Whamcloud - gitweb
lu: add LCT_SESSION context tag for per-request data
[fs/lustre-release.git] / lustre / ptlrpc / service.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  Copyright (C) 2002 Cluster File Systems, Inc.
5  *
6  *   This file is part of the Lustre file system, http://www.lustre.org
7  *   Lustre is a trademark of Cluster File Systems, Inc.
8  *
9  *   You may have signed or agreed to another license before downloading
10  *   this software.  If so, you are bound by the terms and conditions
11  *   of that agreement, and the following does not apply to you.  See the
12  *   LICENSE file included with this distribution for more information.
13  *
14  *   If you did not agree to a different license, then this copy of Lustre
15  *   is open source software; you can redistribute it and/or modify it
16  *   under the terms of version 2 of the GNU General Public License as
17  *   published by the Free Software Foundation.
18  *
19  *   In either case, Lustre is distributed in the hope that it will be
20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
22  *   license text for more details.
23  *
24  */
25
26 #define DEBUG_SUBSYSTEM S_RPC
27 #ifndef __KERNEL__
28 #include <liblustre.h>
29 #include <libcfs/kp30.h>
30 #endif
31 #include <obd_support.h>
32 #include <obd_class.h>
33 #include <lustre_net.h>
34 #include <lu_object.h>
35 #include <lnet/types.h>
36 #include "ptlrpc_internal.h"
37
38 int test_req_buffer_pressure = 0;
39 CFS_MODULE_PARM(test_req_buffer_pressure, "i", int, 0444,
40                 "set non-zero to put pressure on request buffer pools");
41
42 /* forward ref */
43 static int ptlrpc_server_post_idle_rqbds (struct ptlrpc_service *svc);
44
45 static CFS_LIST_HEAD (ptlrpc_all_services);
46 spinlock_t ptlrpc_all_services_lock;
47
48 static char *
49 ptlrpc_alloc_request_buffer (int size)
50 {
51         char *ptr;
52
53         if (size > SVC_BUF_VMALLOC_THRESHOLD)
54                 OBD_VMALLOC(ptr, size);
55         else
56                 OBD_ALLOC(ptr, size);
57
58         return (ptr);
59 }
60
61 static void
62 ptlrpc_free_request_buffer (char *ptr, int size)
63 {
64         if (size > SVC_BUF_VMALLOC_THRESHOLD)
65                 OBD_VFREE(ptr, size);
66         else
67                 OBD_FREE(ptr, size);
68 }
69
70 struct ptlrpc_request_buffer_desc *
71 ptlrpc_alloc_rqbd (struct ptlrpc_service *svc)
72 {
73         struct ptlrpc_request_buffer_desc *rqbd;
74
75         OBD_ALLOC(rqbd, sizeof (*rqbd));
76         if (rqbd == NULL)
77                 return (NULL);
78
79         rqbd->rqbd_service = svc;
80         rqbd->rqbd_refcount = 0;
81         rqbd->rqbd_cbid.cbid_fn = request_in_callback;
82         rqbd->rqbd_cbid.cbid_arg = rqbd;
83         CFS_INIT_LIST_HEAD(&rqbd->rqbd_reqs);
84         rqbd->rqbd_buffer = ptlrpc_alloc_request_buffer(svc->srv_buf_size);
85
86         if (rqbd->rqbd_buffer == NULL) {
87                 OBD_FREE(rqbd, sizeof (*rqbd));
88                 return (NULL);
89         }
90
91         spin_lock(&svc->srv_lock);
92         list_add(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
93         svc->srv_nbufs++;
94         spin_unlock(&svc->srv_lock);
95
96         return (rqbd);
97 }
98
99 void
100 ptlrpc_free_rqbd (struct ptlrpc_request_buffer_desc *rqbd)
101 {
102         struct ptlrpc_service *svc = rqbd->rqbd_service;
103
104         LASSERT (rqbd->rqbd_refcount == 0);
105         LASSERT (list_empty(&rqbd->rqbd_reqs));
106
107         spin_lock(&svc->srv_lock);
108         list_del(&rqbd->rqbd_list);
109         svc->srv_nbufs--;
110         spin_unlock(&svc->srv_lock);
111
112         ptlrpc_free_request_buffer (rqbd->rqbd_buffer, svc->srv_buf_size);
113         OBD_FREE (rqbd, sizeof (*rqbd));
114 }
115
116 int
117 ptlrpc_grow_req_bufs(struct ptlrpc_service *svc)
118 {
119         struct ptlrpc_request_buffer_desc *rqbd;
120         int                                i;
121
122         CDEBUG(D_RPCTRACE, "%s: allocate %d new %d-byte reqbufs (%d/%d left)\n",
123                svc->srv_name, svc->srv_nbuf_per_group, svc->srv_buf_size,
124                svc->srv_nrqbd_receiving, svc->srv_nbufs);
125         for (i = 0; i < svc->srv_nbuf_per_group; i++) {
126                 rqbd = ptlrpc_alloc_rqbd(svc);
127
128                 if (rqbd == NULL) {
129                         CERROR ("%s: Can't allocate request buffer\n",
130                                 svc->srv_name);
131                         return (-ENOMEM);
132                 }
133
134                 if (ptlrpc_server_post_idle_rqbds(svc) < 0)
135                         return (-EAGAIN);
136         }
137
138         return (0);
139 }
140
141 void
142 ptlrpc_save_lock (struct ptlrpc_request *req,
143                   struct lustre_handle *lock, int mode)
144 {
145         struct ptlrpc_reply_state *rs = req->rq_reply_state;
146         int                        idx;
147
148         LASSERT(rs != NULL);
149         LASSERT(rs->rs_nlocks < RS_MAX_LOCKS);
150
151         idx = rs->rs_nlocks++;
152         rs->rs_locks[idx] = *lock;
153         rs->rs_modes[idx] = mode;
154         rs->rs_difficult = 1;
155 }
156
157 void
158 ptlrpc_schedule_difficult_reply (struct ptlrpc_reply_state *rs)
159 {
160         struct ptlrpc_service *svc = rs->rs_service;
161
162 #ifdef CONFIG_SMP
163         LASSERT (spin_is_locked (&svc->srv_lock));
164 #endif
165         LASSERT (rs->rs_difficult);
166         rs->rs_scheduled_ever = 1;              /* flag any notification attempt */
167
168         if (rs->rs_scheduled)                   /* being set up or already notified */
169                 return;
170
171         rs->rs_scheduled = 1;
172         list_del (&rs->rs_list);
173         list_add (&rs->rs_list, &svc->srv_reply_queue);
174         cfs_waitq_signal (&svc->srv_waitq);
175 }
176
177 void
178 ptlrpc_commit_replies (struct obd_device *obd)
179 {
180         struct list_head   *tmp;
181         struct list_head   *nxt;
182
183         /* Find any replies that have been committed and get their service
184          * to attend to complete them. */
185
186         /* CAVEAT EMPTOR: spinlock ordering!!! */
187         spin_lock(&obd->obd_uncommitted_replies_lock);
188
189         list_for_each_safe (tmp, nxt, &obd->obd_uncommitted_replies) {
190                 struct ptlrpc_reply_state *rs =
191                         list_entry(tmp, struct ptlrpc_reply_state, rs_obd_list);
192
193                 LASSERT (rs->rs_difficult);
194
195                 if (rs->rs_transno <= obd->obd_last_committed) {
196                         struct ptlrpc_service *svc = rs->rs_service;
197
198                         spin_lock (&svc->srv_lock);
199                         list_del_init (&rs->rs_obd_list);
200                         ptlrpc_schedule_difficult_reply (rs);
201                         spin_unlock (&svc->srv_lock);
202                 }
203         }
204
205         spin_unlock(&obd->obd_uncommitted_replies_lock);
206 }
207
208 static int
209 ptlrpc_server_post_idle_rqbds (struct ptlrpc_service *svc)
210 {
211         struct ptlrpc_request_buffer_desc *rqbd;
212         int                                rc;
213         int                                posted = 0;
214
215         for (;;) {
216                 spin_lock(&svc->srv_lock);
217
218                 if (list_empty (&svc->srv_idle_rqbds)) {
219                         spin_unlock(&svc->srv_lock);
220                         return (posted);
221                 }
222
223                 rqbd = list_entry(svc->srv_idle_rqbds.next,
224                                   struct ptlrpc_request_buffer_desc,
225                                   rqbd_list);
226                 list_del (&rqbd->rqbd_list);
227
228                 /* assume we will post successfully */
229                 svc->srv_nrqbd_receiving++;
230                 list_add (&rqbd->rqbd_list, &svc->srv_active_rqbds);
231
232                 spin_unlock(&svc->srv_lock);
233
234                 rc = ptlrpc_register_rqbd(rqbd);
235                 if (rc != 0)
236                         break;
237
238                 posted = 1;
239         }
240
241         spin_lock(&svc->srv_lock);
242
243         svc->srv_nrqbd_receiving--;
244         list_del(&rqbd->rqbd_list);
245         list_add_tail(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
246
247         /* Don't complain if no request buffers are posted right now; LNET
248          * won't drop requests because we set the portal lazy! */
249
250         spin_unlock(&svc->srv_lock);
251
252         return (-1);
253 }
254
255 struct ptlrpc_service *ptlrpc_init_svc_conf(struct ptlrpc_service_conf *c,
256                                             svc_handler_t h, char *name,
257                                             struct proc_dir_entry *proc_entry,
258                                             svcreq_printfn_t prntfn)
259 {
260         return ptlrpc_init_svc(c->psc_nbufs, c->psc_bufsize,
261                                c->psc_max_req_size, c->psc_max_reply_size,
262                                c->psc_req_portal, c->psc_rep_portal,
263                                c->psc_watchdog_timeout,
264                                h, name, proc_entry,
265                                prntfn, c->psc_num_threads, c->psc_ctx_tags);
266 }
267 EXPORT_SYMBOL(ptlrpc_init_svc_conf);
268
269 struct ptlrpc_service *
270 ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size, int max_reply_size,
271                 int req_portal, int rep_portal, int watchdog_timeout,
272                 svc_handler_t handler, char *name,
273                 cfs_proc_dir_entry_t *proc_entry,
274                 svcreq_printfn_t svcreq_printfn, int num_threads,
275                 __u32 ctx_tags)
276 {
277         int                    rc;
278         struct ptlrpc_service *service;
279         ENTRY;
280
281         LASSERT (nbufs > 0);
282         LASSERT (bufsize >= max_req_size + SPTLRPC_MAX_PAYLOAD);
283         LASSERT (ctx_tags != 0);
284
285         OBD_ALLOC(service, sizeof(*service));
286         if (service == NULL)
287                 RETURN(NULL);
288
289         /* First initialise enough for early teardown */
290
291         service->srv_name = name;
292         spin_lock_init(&service->srv_lock);
293         CFS_INIT_LIST_HEAD(&service->srv_threads);
294         cfs_waitq_init(&service->srv_waitq);
295
296         service->srv_nbuf_per_group = test_req_buffer_pressure ? 1 : nbufs;
297         service->srv_max_req_size = max_req_size + SPTLRPC_MAX_PAYLOAD;
298         service->srv_buf_size = bufsize;
299         service->srv_rep_portal = rep_portal;
300         service->srv_req_portal = req_portal;
301         service->srv_watchdog_timeout = watchdog_timeout;
302         service->srv_handler = handler;
303         service->srv_request_history_print_fn = svcreq_printfn;
304         service->srv_request_seq = 1;           /* valid seq #s start at 1 */
305         service->srv_request_max_cull_seq = 0;
306         service->srv_num_threads = num_threads;
307         service->srv_ctx_tags = ctx_tags;
308
309         rc = LNetSetLazyPortal(service->srv_req_portal);
310         LASSERT (rc == 0);
311
312         CFS_INIT_LIST_HEAD(&service->srv_request_queue);
313         CFS_INIT_LIST_HEAD(&service->srv_idle_rqbds);
314         CFS_INIT_LIST_HEAD(&service->srv_active_rqbds);
315         CFS_INIT_LIST_HEAD(&service->srv_history_rqbds);
316         CFS_INIT_LIST_HEAD(&service->srv_request_history);
317         CFS_INIT_LIST_HEAD(&service->srv_active_replies);
318         CFS_INIT_LIST_HEAD(&service->srv_reply_queue);
319         CFS_INIT_LIST_HEAD(&service->srv_free_rs_list);
320         cfs_waitq_init(&service->srv_free_rs_waitq);
321
322         spin_lock (&ptlrpc_all_services_lock);
323         list_add (&service->srv_list, &ptlrpc_all_services);
324         spin_unlock (&ptlrpc_all_services_lock);
325
326         /* Now allocate the request buffers */
327         rc = ptlrpc_grow_req_bufs(service);
328         /* We shouldn't be under memory pressure at startup, so
329          * fail if we can't post all our buffers at this time. */
330         if (rc != 0)
331                 GOTO(failed, NULL);
332
333         /* Now allocate pool of reply buffers */
334         /* Increase max reply size to next power of two */
335         service->srv_max_reply_size = 1;
336         while (service->srv_max_reply_size <
337                max_reply_size + SPTLRPC_MAX_PAYLOAD)
338                 service->srv_max_reply_size <<= 1;
339
340         if (proc_entry != NULL)
341                 ptlrpc_lprocfs_register_service(proc_entry, service);
342
343         CDEBUG(D_NET, "%s: Started, listening on portal %d\n",
344                service->srv_name, service->srv_req_portal);
345
346         RETURN(service);
347 failed:
348         ptlrpc_unregister_service(service);
349         return NULL;
350 }
351
352 static void __ptlrpc_server_free_request(struct ptlrpc_request *req)
353 {
354         struct ptlrpc_request_buffer_desc *rqbd = req->rq_rqbd;
355
356         list_del(&req->rq_list);
357
358         if (req->rq_reply_state != NULL) {
359                 ptlrpc_rs_decref(req->rq_reply_state);
360                 req->rq_reply_state = NULL;
361         }
362
363         sptlrpc_svc_ctx_decref(req);
364
365         if (req != &rqbd->rqbd_req) {
366                 /* NB request buffers use an embedded
367                  * req if the incoming req unlinked the
368                  * MD; this isn't one of them! */
369                 OBD_FREE(req, sizeof(*req));
370         }
371 }
372
373 static void
374 ptlrpc_server_free_request(struct ptlrpc_request *req)
375 {
376         struct ptlrpc_request_buffer_desc *rqbd = req->rq_rqbd;
377         struct ptlrpc_service             *svc = rqbd->rqbd_service;
378         int                                refcount;
379         struct list_head                  *tmp;
380         struct list_head                  *nxt;
381
382         spin_lock(&svc->srv_lock);
383
384         svc->srv_n_active_reqs--;
385         list_add(&req->rq_list, &rqbd->rqbd_reqs);
386
387         refcount = --(rqbd->rqbd_refcount);
388         if (refcount == 0) {
389                 /* request buffer is now idle: add to history */
390                 list_del(&rqbd->rqbd_list);
391                 list_add_tail(&rqbd->rqbd_list, &svc->srv_history_rqbds);
392                 svc->srv_n_history_rqbds++;
393
394                 /* cull some history?
395                  * I expect only about 1 or 2 rqbds need to be recycled here */
396                 while (svc->srv_n_history_rqbds > svc->srv_max_history_rqbds) {
397                         rqbd = list_entry(svc->srv_history_rqbds.next,
398                                           struct ptlrpc_request_buffer_desc,
399                                           rqbd_list);
400
401                         list_del(&rqbd->rqbd_list);
402                         svc->srv_n_history_rqbds--;
403
404                         /* remove rqbd's reqs from svc's req history while
405                          * I've got the service lock */
406                         list_for_each(tmp, &rqbd->rqbd_reqs) {
407                                 req = list_entry(tmp, struct ptlrpc_request,
408                                                  rq_list);
409                                 /* Track the highest culled req seq */
410                                 if (req->rq_history_seq >
411                                     svc->srv_request_max_cull_seq)
412                                         svc->srv_request_max_cull_seq =
413                                                 req->rq_history_seq;
414                                 list_del(&req->rq_history_list);
415                         }
416
417                         spin_unlock(&svc->srv_lock);
418
419                         list_for_each_safe(tmp, nxt, &rqbd->rqbd_reqs) {
420                                 req = list_entry(rqbd->rqbd_reqs.next,
421                                                  struct ptlrpc_request,
422                                                  rq_list);
423                                 __ptlrpc_server_free_request(req);
424                         }
425
426                         spin_lock(&svc->srv_lock);
427
428                         /* schedule request buffer for re-use.
429                          * NB I can only do this after I've disposed of their
430                          * reqs; particularly the embedded req */
431                         list_add_tail(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
432                 }
433         } else if (req->rq_reply_state && req->rq_reply_state->rs_prealloc) {
434                  /* If we are low on memory, we are not interested in
435                     history */
436                 list_del(&req->rq_history_list);
437                 __ptlrpc_server_free_request(req);
438         }
439
440         spin_unlock(&svc->srv_lock);
441
442 }
443
444 /* This function makes sure dead exports are evicted in a timely manner.
445    This function is only called when some export receives a message (i.e.,
446    the network is up.) */
447 static void ptlrpc_update_export_timer(struct obd_export *exp, long extra_delay)
448 {
449         struct obd_export *oldest_exp;
450         time_t oldest_time;
451
452         ENTRY;
453
454         LASSERT(exp);
455
456         /* Compensate for slow machines, etc, by faking our request time
457            into the future.  Although this can break the strict time-ordering
458            of the list, we can be really lazy here - we don't have to evict
459            at the exact right moment.  Eventually, all silent exports
460            will make it to the top of the list. */
461         exp->exp_last_request_time = max(exp->exp_last_request_time,
462                                          (time_t)CURRENT_SECONDS + extra_delay);
463
464         CDEBUG(D_INFO, "updating export %s at %ld\n",
465                exp->exp_client_uuid.uuid,
466                exp->exp_last_request_time);
467
468         /* exports may get disconnected from the chain even though the
469            export has references, so we must keep the spin lock while
470            manipulating the lists */
471         spin_lock(&exp->exp_obd->obd_dev_lock);
472
473         if (list_empty(&exp->exp_obd_chain_timed)) {
474                 /* this one is not timed */
475                 spin_unlock(&exp->exp_obd->obd_dev_lock);
476                 EXIT;
477                 return;
478         }
479
480         list_move_tail(&exp->exp_obd_chain_timed,
481                        &exp->exp_obd->obd_exports_timed);
482
483         oldest_exp = list_entry(exp->exp_obd->obd_exports_timed.next,
484                                 struct obd_export, exp_obd_chain_timed);
485         oldest_time = oldest_exp->exp_last_request_time;
486         spin_unlock(&exp->exp_obd->obd_dev_lock);
487
488         if (exp->exp_obd->obd_recovering) {
489                 /* be nice to everyone during recovery */
490                 EXIT;
491                 return;
492         }
493
494         /* Note - racing to start/reset the obd_eviction timer is safe */
495         if (exp->exp_obd->obd_eviction_timer == 0) {
496                 /* Check if the oldest entry is expired. */
497                 if (CURRENT_SECONDS > (oldest_time +
498                                        (3 * obd_timeout / 2) + extra_delay)) {
499                         /* We need a second timer, in case the net was down and
500                          * it just came back. Since the pinger may skip every
501                          * other PING_INTERVAL (see note in ptlrpc_pinger_main),
502                          * we better wait for 3. */
503                         exp->exp_obd->obd_eviction_timer = CURRENT_SECONDS +
504                                 3 * PING_INTERVAL;
505                         CDEBUG(D_HA, "%s: Think about evicting %s from %ld\n",
506                                exp->exp_obd->obd_name, obd_export_nid2str(exp),
507                                oldest_time);
508                 }
509         } else {
510                 if (CURRENT_SECONDS > (exp->exp_obd->obd_eviction_timer +
511                                        extra_delay)) {
512                         /* The evictor won't evict anyone who we've heard from
513                          * recently, so we don't have to check before we start
514                          * it. */
515                         if (!ping_evictor_wake(exp))
516                                 exp->exp_obd->obd_eviction_timer = 0;
517                 }
518         }
519
520         EXIT;
521 }
522
523 #ifndef __KERNEL__
524 int lu_context_init(struct lu_context *ctx, __u32 tags)
525 {
526         return 0;
527 }
528
529 void lu_context_fini(struct lu_context *ctx)
530 {
531 }
532
533 void lu_context_enter(struct lu_context *ctx)
534 {
535 }
536
537 void lu_context_exit(struct lu_context *ctx)
538 {
539 }
540
541 #endif
542
543 static int
544 ptlrpc_server_handle_request(struct ptlrpc_service *svc,
545                              struct ptlrpc_thread *thread)
546 {
547         struct ptlrpc_request *request;
548         struct timeval         work_start;
549         struct timeval         work_end;
550         long                   timediff;
551         int                    rc;
552         ENTRY;
553
554         LASSERT(svc);
555
556         spin_lock(&svc->srv_lock);
557         if (list_empty (&svc->srv_request_queue) ||
558             (svc->srv_n_difficult_replies != 0 &&
559              svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
560                 /* If all the other threads are handling requests, I must
561                  * remain free to handle any 'difficult' reply that might
562                  * block them */
563                 spin_unlock(&svc->srv_lock);
564                 RETURN(0);
565         }
566
567         request = list_entry (svc->srv_request_queue.next,
568                               struct ptlrpc_request, rq_list);
569         list_del_init (&request->rq_list);
570         svc->srv_n_queued_reqs--;
571         svc->srv_n_active_reqs++;
572
573         spin_unlock(&svc->srv_lock);
574
575         do_gettimeofday(&work_start);
576         timediff = cfs_timeval_sub(&work_start, &request->rq_arrival_time,NULL);
577         if (svc->srv_stats != NULL) {
578                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
579                                     timediff);
580                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
581                                     svc->srv_n_queued_reqs);
582                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
583                                     svc->srv_n_active_reqs);
584         }
585
586         rc = sptlrpc_svc_unwrap_request(request);
587         switch (rc) {
588         case SECSVC_OK:
589                 break;
590         case SECSVC_COMPLETE:
591                 target_send_reply(request, 0, OBD_FAIL_MDS_ALL_REPLY_NET);
592                 goto put_conn;
593         case SECSVC_DROP:
594                 goto out;
595         default:
596                 LBUG();
597         }
598
599 #if SWAB_PARANOIA
600         /* Clear request swab mask; this is a new request */
601         request->rq_req_swab_mask = 0;
602 #endif
603         rc = lustre_unpack_msg(request->rq_reqmsg, request->rq_reqlen);
604         if (rc != 0) {
605                 CERROR ("error unpacking request: ptl %d from %s"
606                         " xid "LPU64"\n", svc->srv_req_portal,
607                         libcfs_id2str(request->rq_peer), request->rq_xid);
608                 goto out;
609         }
610
611         rc = lustre_unpack_ptlrpc_body(request->rq_reqmsg);
612         if (rc) {
613                 CERROR ("error unpacking ptlrpc body: ptl %d from %s"
614                         " xid "LPU64"\n", svc->srv_req_portal,
615                         libcfs_id2str(request->rq_peer), request->rq_xid);
616                 goto out;
617         }
618
619         rc = -EINVAL;
620         if (lustre_msg_get_type(request->rq_reqmsg) != PTL_RPC_MSG_REQUEST) {
621                 CERROR("wrong packet type received (type=%u) from %s\n",
622                        lustre_msg_get_type(request->rq_reqmsg),
623                        libcfs_id2str(request->rq_peer));
624                 goto out;
625         }
626
627         rc = lu_context_init(&request->rq_session, LCT_SESSION);
628         if (rc) {
629                 CERROR("Failure to initialize session: %d\n", rc);
630                 goto out;
631         }
632         request->rq_session.lc_thread = thread;
633         lu_context_enter(&request->rq_session);
634
635         CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
636
637         request->rq_svc_thread = thread;
638         request->rq_export = class_conn2export(
639                                      lustre_msg_get_handle(request->rq_reqmsg));
640
641         if (request->rq_export) {
642                 if (lustre_msg_get_conn_cnt(request->rq_reqmsg) <
643                     request->rq_export->exp_conn_cnt) {
644                         DEBUG_REQ(D_ERROR, request,
645                                   "DROPPING req from old connection %d < %d",
646                                   lustre_msg_get_conn_cnt(request->rq_reqmsg),
647                                   request->rq_export->exp_conn_cnt);
648                         goto put_conn;
649                 }
650                 if (request->rq_export->exp_obd &&
651                     request->rq_export->exp_obd->obd_fail) {
652                         /* Failing over, don't handle any more reqs, send
653                            error response instead. */
654                         CDEBUG(D_HA, "Dropping req %p for failed obd %s\n",
655                                request, request->rq_export->exp_obd->obd_name);
656                         request->rq_status = -ENODEV;
657                         ptlrpc_error(request);
658                         goto put_conn;
659                 }
660
661                 ptlrpc_update_export_timer(request->rq_export, timediff/500000);
662         }
663
664         /* Discard requests queued for longer than my timeout.  If the
665          * client's timeout is similar to mine, she'll be timing out this
666          * REQ anyway (bug 1502) */
667         if (timediff / 1000000 > (long)obd_timeout) {
668                 CERROR("Dropping timed-out opc %d request from %s"
669                        ": %ld seconds old\n",
670                        lustre_msg_get_opc(request->rq_reqmsg),
671                        libcfs_id2str(request->rq_peer),
672                        timediff / 1000000);
673                 goto put_conn;
674         }
675
676         request->rq_phase = RQ_PHASE_INTERPRET;
677
678         CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:nid:opc "
679                "%s:%s+%d:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
680                (request->rq_export ?
681                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
682                (request->rq_export ?
683                 atomic_read(&request->rq_export->exp_refcount) : -99),
684                lustre_msg_get_status(request->rq_reqmsg), request->rq_xid,
685                libcfs_id2str(request->rq_peer),
686                lustre_msg_get_opc(request->rq_reqmsg));
687
688         rc = svc->srv_handler(request);
689
690         request->rq_phase = RQ_PHASE_COMPLETE;
691
692         CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:nid:opc "
693                "%s:%s+%d:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
694                (request->rq_export ?
695                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
696                (request->rq_export ?
697                 atomic_read(&request->rq_export->exp_refcount) : -99),
698                lustre_msg_get_status(request->rq_reqmsg), request->rq_xid,
699                libcfs_id2str(request->rq_peer),
700                lustre_msg_get_opc(request->rq_reqmsg));
701
702 put_conn:
703         if (request->rq_export != NULL)
704                 class_export_put(request->rq_export);
705
706  out:
707         lu_context_exit(&request->rq_session);
708         lu_context_fini(&request->rq_session);
709
710         do_gettimeofday(&work_end);
711
712         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
713
714         if (timediff / 1000000 > (long)obd_timeout)
715                 CERROR("request "LPU64" opc %u from %s processed in %lds "
716                        "trans "LPU64" rc %d/%d\n",
717                        request->rq_xid,
718                        request->rq_reqmsg ?
719                                 lustre_msg_get_opc(request->rq_reqmsg) : 0,
720                        libcfs_id2str(request->rq_peer),
721                        cfs_timeval_sub(&work_end, &request->rq_arrival_time,
722                                        NULL) / 1000000,
723                        request->rq_repmsg ?
724                                 lustre_msg_get_transno(request->rq_repmsg) :
725                                 request->rq_transno, request->rq_status,
726                        request->rq_repmsg ?
727                                 lustre_msg_get_status(request->rq_repmsg) :
728                                  -999);
729         else
730                 CDEBUG(D_HA, "request "LPU64" opc %u from %s processed in "
731                        "%ldus (%ldus total) trans "LPU64" rc %d/%d\n",
732                        request->rq_xid,
733                        request->rq_reqmsg ?
734                                 lustre_msg_get_opc(request->rq_reqmsg) : 0,
735                        libcfs_id2str(request->rq_peer), timediff,
736                        cfs_timeval_sub(&work_end, &request->rq_arrival_time,
737                                        NULL),
738                        request->rq_transno, request->rq_status,
739                        request->rq_repmsg ?
740                                 lustre_msg_get_status(request->rq_repmsg) :
741                                 -999);
742
743         if (svc->srv_stats != NULL && request->rq_reqmsg != NULL) {
744                 int opc = opcode_offset(lustre_msg_get_opc(request->rq_reqmsg));
745                 if (opc > 0) {
746                         LASSERT(opc < LUSTRE_MAX_OPCODES);
747                         lprocfs_counter_add(svc->srv_stats,
748                                             opc + PTLRPC_LAST_CNTR,
749                                             timediff);
750                 }
751         }
752
753         ptlrpc_server_free_request(request);
754
755         RETURN(1);
756 }
757
758 static int
759 ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
760 {
761         struct ptlrpc_reply_state *rs;
762         struct obd_export         *exp;
763         struct obd_device         *obd;
764         int                        nlocks;
765         int                        been_handled;
766         ENTRY;
767
768         spin_lock(&svc->srv_lock);
769         if (list_empty (&svc->srv_reply_queue)) {
770                 spin_unlock(&svc->srv_lock);
771                 RETURN(0);
772         }
773
774         rs = list_entry (svc->srv_reply_queue.next,
775                          struct ptlrpc_reply_state, rs_list);
776
777         exp = rs->rs_export;
778         obd = exp->exp_obd;
779
780         LASSERT (rs->rs_difficult);
781         LASSERT (rs->rs_scheduled);
782
783         list_del_init (&rs->rs_list);
784
785         /* Disengage from notifiers carefully (lock order - irqrestore below!)*/
786         spin_unlock(&svc->srv_lock);
787
788         spin_lock (&obd->obd_uncommitted_replies_lock);
789         /* Noop if removed already */
790         list_del_init (&rs->rs_obd_list);
791         spin_unlock (&obd->obd_uncommitted_replies_lock);
792
793         spin_lock (&exp->exp_lock);
794         /* Noop if removed already */
795         list_del_init (&rs->rs_exp_list);
796         spin_unlock (&exp->exp_lock);
797
798         spin_lock(&svc->srv_lock);
799
800         been_handled = rs->rs_handled;
801         rs->rs_handled = 1;
802
803         nlocks = rs->rs_nlocks;                 /* atomic "steal", but */
804         rs->rs_nlocks = 0;                      /* locks still on rs_locks! */
805
806         if (nlocks == 0 && !been_handled) {
807                 /* If we see this, we should already have seen the warning
808                  * in mds_steal_ack_locks()  */
809                 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
810                       " o%d NID %s\n",
811                       rs,
812                       rs->rs_xid, rs->rs_transno,
813                       lustre_msg_get_opc(rs->rs_msg),
814                       libcfs_nid2str(exp->exp_connection->c_peer.nid));
815         }
816
817         if ((!been_handled && rs->rs_on_net) || nlocks > 0) {
818                 spin_unlock(&svc->srv_lock);
819
820                 if (!been_handled && rs->rs_on_net) {
821                         LNetMDUnlink(rs->rs_md_h);
822                         /* Ignore return code; we're racing with
823                          * completion... */
824                 }
825
826                 while (nlocks-- > 0)
827                         ldlm_lock_decref(&rs->rs_locks[nlocks],
828                                          rs->rs_modes[nlocks]);
829
830                 spin_lock(&svc->srv_lock);
831         }
832
833         rs->rs_scheduled = 0;
834
835         if (!rs->rs_on_net) {
836                 /* Off the net */
837                 svc->srv_n_difficult_replies--;
838                 spin_unlock(&svc->srv_lock);
839
840                 class_export_put (exp);
841                 rs->rs_export = NULL;
842                 ptlrpc_rs_decref (rs);
843                 atomic_dec (&svc->srv_outstanding_replies);
844                 RETURN(1);
845         }
846
847         /* still on the net; callback will schedule */
848         spin_unlock(&svc->srv_lock);
849         RETURN(1);
850 }
851
852 #ifndef __KERNEL__
853 /* FIXME make use of timeout later */
854 int
855 liblustre_check_services (void *arg)
856 {
857         int  did_something = 0;
858         int  rc;
859         struct list_head *tmp, *nxt;
860         ENTRY;
861
862         /* I'm relying on being single threaded, not to have to lock
863          * ptlrpc_all_services etc */
864         list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
865                 struct ptlrpc_service *svc =
866                         list_entry (tmp, struct ptlrpc_service, srv_list);
867
868                 if (svc->srv_nthreads != 0)     /* I've recursed */
869                         continue;
870
871                 /* service threads can block for bulk, so this limits us
872                  * (arbitrarily) to recursing 1 stack frame per service.
873                  * Note that the problem with recursion is that we have to
874                  * unwind completely before our caller can resume. */
875
876                 svc->srv_nthreads++;
877
878                 do {
879                         rc = ptlrpc_server_handle_reply(svc);
880                         rc |= ptlrpc_server_handle_request(svc, NULL);
881                         rc |= (ptlrpc_server_post_idle_rqbds(svc) > 0);
882                         did_something |= rc;
883                 } while (rc);
884
885                 svc->srv_nthreads--;
886         }
887
888         RETURN(did_something);
889 }
890 #define ptlrpc_stop_all_threads(s) do {} while (0)
891
892 #else /* __KERNEL__ */
893
894 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
895 void ptlrpc_daemonize(char *name)
896 {
897         struct fs_struct *fs = current->fs;
898
899         atomic_inc(&fs->count);
900         cfs_daemonize(name);
901         exit_fs(cfs_current());
902         current->fs = fs;
903         ll_set_fs_pwd(current->fs, init_task.fs->pwdmnt, init_task.fs->pwd);
904 }
905
906 static void
907 ptlrpc_check_rqbd_pool(struct ptlrpc_service *svc)
908 {
909         int avail = svc->srv_nrqbd_receiving;
910         int low_water = test_req_buffer_pressure ? 0 :
911                         svc->srv_nbuf_per_group/2;
912
913         /* NB I'm not locking; just looking. */
914
915         /* CAVEAT EMPTOR: We might be allocating buffers here because we've
916          * allowed the request history to grow out of control.  We could put a
917          * sanity check on that here and cull some history if we need the
918          * space. */
919
920         if (avail <= low_water)
921                 ptlrpc_grow_req_bufs(svc);
922
923         if (svc->srv_stats)
924                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQBUF_AVAIL_CNTR,
925                                     avail);
926 }
927
928 static int
929 ptlrpc_retry_rqbds(void *arg)
930 {
931         struct ptlrpc_service *svc = (struct ptlrpc_service *)arg;
932
933         svc->srv_rqbd_timeout = 0;
934         return (-ETIMEDOUT);
935 }
936
937 static int ptlrpc_main(void *arg)
938 {
939         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
940         struct ptlrpc_service  *svc = data->svc;
941         struct ptlrpc_thread   *thread = data->thread;
942         struct ptlrpc_reply_state *rs;
943         struct lc_watchdog     *watchdog;
944 #ifdef WITH_GROUP_INFO
945         struct group_info *ginfo = NULL;
946 #endif
947         struct lu_context ctx;
948         int rc = 0;
949         ENTRY;
950
951         ptlrpc_daemonize(data->name);
952
953 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,9) && CONFIG_NUMA
954         /* we need to do this before any per-thread allocation is done so that
955          * we get the per-thread allocations on local node.  bug 7342 */
956         if (svc->srv_cpu_affinity) {
957                 int cpu, num_cpu;
958
959                 for (cpu = 0, num_cpu = 0; cpu < NR_CPUS; cpu++) {
960                         if (!cpu_online(cpu))
961                                 continue;
962                         if (num_cpu == thread->t_id % num_online_cpus())
963                                 break;
964                         num_cpu++;
965                 }
966                 set_cpus_allowed(cfs_current(), node_to_cpumask(cpu_to_node(cpu)));
967         }
968 #endif
969
970 #ifdef WITH_GROUP_INFO
971         ginfo = groups_alloc(0);
972         if (!ginfo) {
973                 rc = -ENOMEM;
974                 goto out;
975         }
976
977         set_current_groups(ginfo);
978         put_group_info(ginfo);
979 #endif
980
981         if (svc->srv_init != NULL) {
982                 rc = svc->srv_init(thread);
983                 if (rc)
984                         goto out;
985         }
986
987         rc = lu_context_init(&ctx, svc->srv_ctx_tags);
988         if (rc)
989                 goto out_srv_init;
990
991         thread->t_ctx = &ctx;
992         ctx.lc_thread = thread;
993
994         /* Alloc reply state structure for this one */
995         OBD_ALLOC_GFP(rs, svc->srv_max_reply_size, CFS_ALLOC_STD);
996         if (!rs) {
997                 rc = -ENOMEM;
998                 goto out_srv_init;
999         }
1000
1001         /* Record that the thread is running */
1002         thread->t_flags = SVC_RUNNING;
1003         /*
1004          * wake up our creator. Note: @data is invalid after this point,
1005          * because it's allocated on ptlrpc_start_thread() stack.
1006          */
1007         cfs_waitq_signal(&thread->t_ctl_waitq);
1008
1009         watchdog = lc_watchdog_add(svc->srv_watchdog_timeout, NULL, NULL);
1010
1011         spin_lock(&svc->srv_lock);
1012         svc->srv_nthreads++;
1013         list_add(&rs->rs_list, &svc->srv_free_rs_list);
1014         spin_unlock(&svc->srv_lock);
1015         cfs_waitq_signal(&svc->srv_free_rs_waitq);
1016
1017         CDEBUG(D_NET, "service thread %d started\n", thread->t_id);
1018
1019         /* XXX maintain a list of all managed devices: insert here */
1020
1021         while ((thread->t_flags & SVC_STOPPING) == 0 ||
1022                svc->srv_n_difficult_replies != 0) {
1023                 /* Don't exit while there are replies to be handled */
1024                 struct l_wait_info lwi = LWI_TIMEOUT(svc->srv_rqbd_timeout,
1025                                                      ptlrpc_retry_rqbds, svc);
1026
1027                 lc_watchdog_disable(watchdog);
1028
1029                 l_wait_event_exclusive (svc->srv_waitq,
1030                               ((thread->t_flags & SVC_STOPPING) != 0 &&
1031                                svc->srv_n_difficult_replies == 0) ||
1032                               (!list_empty(&svc->srv_idle_rqbds) &&
1033                                svc->srv_rqbd_timeout == 0) ||
1034                               !list_empty (&svc->srv_reply_queue) ||
1035                               (!list_empty (&svc->srv_request_queue) &&
1036                                (svc->srv_n_difficult_replies == 0 ||
1037                                 svc->srv_n_active_reqs <
1038                                 (svc->srv_nthreads - 1))),
1039                               &lwi);
1040
1041                 lc_watchdog_touch(watchdog);
1042
1043                 ptlrpc_check_rqbd_pool(svc);
1044
1045                 if (!list_empty (&svc->srv_reply_queue))
1046                         ptlrpc_server_handle_reply (svc);
1047
1048                 /* only handle requests if there are no difficult replies
1049                  * outstanding, or I'm not the last thread handling
1050                  * requests */
1051                 if (!list_empty (&svc->srv_request_queue) &&
1052                     (svc->srv_n_difficult_replies == 0 ||
1053                      svc->srv_n_active_reqs < (svc->srv_nthreads - 1))) {
1054                         lu_context_enter(&ctx);
1055                         ptlrpc_server_handle_request(svc, thread);
1056                         lu_context_exit(&ctx);
1057                 }
1058
1059                 if (!list_empty(&svc->srv_idle_rqbds) &&
1060                     ptlrpc_server_post_idle_rqbds(svc) < 0) {
1061                         /* I just failed to repost request buffers.  Wait
1062                          * for a timeout (unless something else happens)
1063                          * before I try again */
1064                         svc->srv_rqbd_timeout = cfs_time_seconds(1)/10;
1065                         CDEBUG(D_RPCTRACE,"Posted buffers: %d\n",
1066                                svc->srv_nrqbd_receiving);
1067                 }
1068         }
1069
1070         lc_watchdog_delete(watchdog);
1071
1072 out_srv_init:
1073         /*
1074          * deconstruct service specific state created by ptlrpc_start_thread()
1075          */
1076         if (svc->srv_done != NULL)
1077                 svc->srv_done(thread);
1078
1079 out:
1080         lu_context_fini(&ctx);
1081
1082         CDEBUG(D_NET, "service thread %d exiting: rc %d\n", thread->t_id, rc);
1083
1084         spin_lock(&svc->srv_lock);
1085         svc->srv_nthreads--;                    /* must know immediately */
1086         thread->t_id = rc;
1087         thread->t_flags = SVC_STOPPED;
1088
1089         cfs_waitq_signal(&thread->t_ctl_waitq);
1090         spin_unlock(&svc->srv_lock);
1091
1092         return rc;
1093 }
1094
1095 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
1096                                struct ptlrpc_thread *thread)
1097 {
1098         struct l_wait_info lwi = { 0 };
1099
1100         spin_lock(&svc->srv_lock);
1101         thread->t_flags = SVC_STOPPING;
1102         spin_unlock(&svc->srv_lock);
1103
1104         cfs_waitq_broadcast(&svc->srv_waitq);
1105         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
1106                      &lwi);
1107
1108         spin_lock(&svc->srv_lock);
1109         list_del(&thread->t_link);
1110         spin_unlock(&svc->srv_lock);
1111
1112         OBD_FREE(thread, sizeof(*thread));
1113 }
1114
1115 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
1116 {
1117         struct ptlrpc_thread *thread;
1118
1119         spin_lock(&svc->srv_lock);
1120         while (!list_empty(&svc->srv_threads)) {
1121                 thread = list_entry(svc->srv_threads.next,
1122                                     struct ptlrpc_thread, t_link);
1123
1124                 spin_unlock(&svc->srv_lock);
1125                 ptlrpc_stop_thread(svc, thread);
1126                 spin_lock(&svc->srv_lock);
1127         }
1128
1129         spin_unlock(&svc->srv_lock);
1130 }
1131
1132 /* @base_name should be 11 characters or less - 3 will be added on */
1133 int ptlrpc_start_threads(struct obd_device *dev, struct ptlrpc_service *svc,
1134                          char *base_name)
1135 {
1136         int i, rc = 0;
1137         ENTRY;
1138
1139         LASSERT(svc->srv_num_threads > 0);
1140         for (i = 0; i < svc->srv_num_threads; i++) {
1141                 char name[32];
1142                 sprintf(name, "%s_%02d", base_name, i);
1143                 rc = ptlrpc_start_thread(dev, svc, name, i);
1144                 if (rc) {
1145                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
1146                                i, rc);
1147                         ptlrpc_stop_all_threads(svc);
1148                 }
1149         }
1150         RETURN(rc);
1151 }
1152
1153 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
1154                         char *name, int id)
1155 {
1156         struct l_wait_info lwi = { 0 };
1157         struct ptlrpc_svc_data d;
1158         struct ptlrpc_thread *thread;
1159         int rc;
1160         ENTRY;
1161
1162         OBD_ALLOC(thread, sizeof(*thread));
1163         if (thread == NULL)
1164                 RETURN(-ENOMEM);
1165         cfs_waitq_init(&thread->t_ctl_waitq);
1166         thread->t_id = id;
1167
1168         spin_lock(&svc->srv_lock);
1169         list_add(&thread->t_link, &svc->srv_threads);
1170         spin_unlock(&svc->srv_lock);
1171
1172         d.dev = dev;
1173         d.svc = svc;
1174         d.name = name;
1175         d.thread = thread;
1176
1177         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
1178          * just drop the VM and FILES in ptlrpc_daemonize() right away.
1179          */
1180         rc = cfs_kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
1181         if (rc < 0) {
1182                 CERROR("cannot start thread '%s': rc %d\n", name, rc);
1183
1184                 spin_lock(&svc->srv_lock);
1185                 list_del(&thread->t_link);
1186                 spin_unlock(&svc->srv_lock);
1187
1188                 OBD_FREE(thread, sizeof(*thread));
1189                 RETURN(rc);
1190         }
1191         l_wait_event(thread->t_ctl_waitq,
1192                      thread->t_flags & (SVC_RUNNING | SVC_STOPPED), &lwi);
1193
1194         rc = (thread->t_flags & SVC_STOPPED) ? thread->t_id : 0;
1195         RETURN(rc);
1196 }
1197 #endif
1198
1199 int ptlrpc_unregister_service(struct ptlrpc_service *service)
1200 {
1201         int                   rc;
1202         struct l_wait_info    lwi;
1203         struct list_head     *tmp;
1204         struct ptlrpc_reply_state *rs, *t;
1205
1206         ptlrpc_stop_all_threads(service);
1207         LASSERT(list_empty(&service->srv_threads));
1208
1209         spin_lock (&ptlrpc_all_services_lock);
1210         list_del_init (&service->srv_list);
1211         spin_unlock (&ptlrpc_all_services_lock);
1212
1213         ptlrpc_lprocfs_unregister_service(service);
1214
1215         /* All history will be culled when the next request buffer is
1216          * freed */
1217         service->srv_max_history_rqbds = 0;
1218
1219         CDEBUG(D_NET, "%s: tearing down\n", service->srv_name);
1220
1221         rc = LNetClearLazyPortal(service->srv_req_portal);
1222         LASSERT (rc == 0);
1223
1224         /* Unlink all the request buffers.  This forces a 'final' event with
1225          * its 'unlink' flag set for each posted rqbd */
1226         list_for_each(tmp, &service->srv_active_rqbds) {
1227                 struct ptlrpc_request_buffer_desc *rqbd =
1228                         list_entry(tmp, struct ptlrpc_request_buffer_desc,
1229                                    rqbd_list);
1230
1231                 rc = LNetMDUnlink(rqbd->rqbd_md_h);
1232                 LASSERT (rc == 0 || rc == -ENOENT);
1233         }
1234
1235         /* Wait for the network to release any buffers it's currently
1236          * filling */
1237         for (;;) {
1238                 spin_lock(&service->srv_lock);
1239                 rc = service->srv_nrqbd_receiving;
1240                 spin_unlock(&service->srv_lock);
1241
1242                 if (rc == 0)
1243                         break;
1244
1245                 /* Network access will complete in finite time but the HUGE
1246                  * timeout lets us CWARN for visibility of sluggish NALs */
1247                 lwi = LWI_TIMEOUT(cfs_time_seconds(300), NULL, NULL);
1248                 rc = l_wait_event(service->srv_waitq,
1249                                   service->srv_nrqbd_receiving == 0,
1250                                   &lwi);
1251                 if (rc == -ETIMEDOUT)
1252                         CWARN("Service %s waiting for request buffers\n",
1253                               service->srv_name);
1254         }
1255
1256         /* schedule all outstanding replies to terminate them */
1257         spin_lock(&service->srv_lock);
1258         while (!list_empty(&service->srv_active_replies)) {
1259                 struct ptlrpc_reply_state *rs =
1260                         list_entry(service->srv_active_replies.next,
1261                                    struct ptlrpc_reply_state, rs_list);
1262                 ptlrpc_schedule_difficult_reply(rs);
1263         }
1264         spin_unlock(&service->srv_lock);
1265
1266         /* purge the request queue.  NB No new replies (rqbds all unlinked)
1267          * and no service threads, so I'm the only thread noodling the
1268          * request queue now */
1269         while (!list_empty(&service->srv_request_queue)) {
1270                 struct ptlrpc_request *req =
1271                         list_entry(service->srv_request_queue.next,
1272                                    struct ptlrpc_request,
1273                                    rq_list);
1274
1275                 list_del(&req->rq_list);
1276                 service->srv_n_queued_reqs--;
1277                 service->srv_n_active_reqs++;
1278
1279                 ptlrpc_server_free_request(req);
1280         }
1281         LASSERT(service->srv_n_queued_reqs == 0);
1282         LASSERT(service->srv_n_active_reqs == 0);
1283         LASSERT(service->srv_n_history_rqbds == 0);
1284         LASSERT(list_empty(&service->srv_active_rqbds));
1285
1286         /* Now free all the request buffers since nothing references them
1287          * any more... */
1288         while (!list_empty(&service->srv_idle_rqbds)) {
1289                 struct ptlrpc_request_buffer_desc *rqbd =
1290                         list_entry(service->srv_idle_rqbds.next,
1291                                    struct ptlrpc_request_buffer_desc,
1292                                    rqbd_list);
1293
1294                 ptlrpc_free_rqbd(rqbd);
1295         }
1296
1297         /* wait for all outstanding replies to complete (they were
1298          * scheduled having been flagged to abort above) */
1299         while (atomic_read(&service->srv_outstanding_replies) != 0) {
1300                 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(10), NULL, NULL);
1301
1302                 rc = l_wait_event(service->srv_waitq,
1303                                   !list_empty(&service->srv_reply_queue), &lwi);
1304                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
1305
1306                 if (rc == 0) {
1307                         ptlrpc_server_handle_reply(service);
1308                         continue;
1309                 }
1310                 CWARN("Unexpectedly long timeout %p\n", service);
1311         }
1312
1313         list_for_each_entry_safe(rs, t, &service->srv_free_rs_list, rs_list) {
1314                 list_del(&rs->rs_list);
1315                 OBD_FREE(rs, service->srv_max_reply_size);
1316         }
1317
1318         OBD_FREE(service, sizeof(*service));
1319         return 0;
1320 }
1321
1322 /* Returns 0 if the service is healthy.
1323  *
1324  * Right now, it just checks to make sure that requests aren't languishing
1325  * in the queue.  We'll use this health check to govern whether a node needs
1326  * to be shot, so it's intentionally non-aggressive. */
1327 int ptlrpc_service_health_check(struct ptlrpc_service *svc)
1328 {
1329         struct ptlrpc_request *request;
1330         struct timeval         right_now;
1331         long                   timediff, cutoff;
1332         int                    rc = 0;
1333
1334         if (svc == NULL)
1335                 return 0;
1336
1337         spin_lock(&svc->srv_lock);
1338
1339         if (list_empty(&svc->srv_request_queue))
1340                 goto out;
1341
1342         request = list_entry(svc->srv_request_queue.next,
1343                              struct ptlrpc_request, rq_list);
1344
1345         do_gettimeofday(&right_now);
1346         timediff = cfs_timeval_sub(&right_now, &request->rq_arrival_time, NULL);
1347
1348         cutoff = obd_health_check_timeout;
1349
1350         if (timediff / 1000000 > cutoff) {
1351                 rc = -1;
1352                 goto out;
1353         }
1354
1355  out:
1356         spin_unlock(&svc->srv_lock);
1357         return rc;
1358 }