Whamcloud - gitweb
d8271f26f0760165b26670f6d29c9055cc28db41
[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 Lustre, http://www.lustre.org.
7  *
8  *   Lustre is free software; you can redistribute it and/or
9  *   modify it under the terms of version 2 of the GNU General Public
10  *   License as published by the Free Software Foundation.
11  *
12  *   Lustre is distributed in the hope that it will be useful,
13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  *   GNU General Public License for more details.
16  *
17  *   You should have received a copy of the GNU General Public License
18  *   along with Lustre; if not, write to the Free Software
19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20  *
21  */
22
23 #define DEBUG_SUBSYSTEM S_RPC
24 #ifndef __KERNEL__
25 #include <liblustre.h>
26 #include <linux/kp30.h>
27 #endif
28 #include <linux/obd_support.h>
29 #include <linux/obd_class.h>
30 #include <linux/lustre_net.h>
31 #include <portals/types.h>
32 #include "ptlrpc_internal.h"
33
34 static LIST_HEAD (ptlrpc_all_services);
35 static spinlock_t ptlrpc_all_services_lock = SPIN_LOCK_UNLOCKED;
36
37 static void
38 ptlrpc_free_server_req (struct ptlrpc_request *req)
39 {
40         /* The last request to be received into a request buffer uses space
41          * in the request buffer descriptor, otherwise requests are
42          * allocated dynamically in the incoming reply event handler */
43         if (req == &req->rq_rqbd->rqbd_req)
44                 return;
45
46         OBD_FREE(req, sizeof(*req));
47 }
48         
49 static char *
50 ptlrpc_alloc_request_buffer (int size)
51 {
52         char *ptr;
53         
54         if (size > SVC_BUF_VMALLOC_THRESHOLD)
55                 OBD_VMALLOC(ptr, size);
56         else
57                 OBD_ALLOC(ptr, size);
58         
59         return (ptr);
60 }
61
62 static void
63 ptlrpc_free_request_buffer (char *ptr, int size)
64 {
65         if (size > SVC_BUF_VMALLOC_THRESHOLD)
66                 OBD_VFREE(ptr, size);
67         else
68                 OBD_FREE(ptr, size);
69 }
70
71 struct ptlrpc_request_buffer_desc *
72 ptlrpc_alloc_rqbd (struct ptlrpc_srv_ni *srv_ni)
73 {
74         struct ptlrpc_service             *svc = srv_ni->sni_service;
75         unsigned long                      flags;
76         struct ptlrpc_request_buffer_desc *rqbd;
77
78         OBD_ALLOC(rqbd, sizeof (*rqbd));
79         if (rqbd == NULL)
80                 return (NULL);
81
82         rqbd->rqbd_srv_ni = srv_ni;
83         rqbd->rqbd_refcount = 0;
84         rqbd->rqbd_cbid.cbid_fn = request_in_callback;
85         rqbd->rqbd_cbid.cbid_arg = rqbd;
86         rqbd->rqbd_buffer = ptlrpc_alloc_request_buffer(svc->srv_buf_size);
87
88         if (rqbd->rqbd_buffer == NULL) {
89                 OBD_FREE(rqbd, sizeof (*rqbd));
90                 return (NULL);
91         }
92
93         spin_lock_irqsave (&svc->srv_lock, flags);
94         list_add(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
95         svc->srv_nbufs++;
96         spin_unlock_irqrestore (&svc->srv_lock, flags);
97
98         return (rqbd);
99 }
100
101 void
102 ptlrpc_free_rqbd (struct ptlrpc_request_buffer_desc *rqbd) 
103 {
104         struct ptlrpc_srv_ni  *sni = rqbd->rqbd_srv_ni;
105         struct ptlrpc_service *svc = sni->sni_service;
106         unsigned long          flags;
107         
108         LASSERT (rqbd->rqbd_refcount == 0);
109
110         spin_lock_irqsave(&svc->srv_lock, flags);
111         list_del(&rqbd->rqbd_list);
112         svc->srv_nbufs--;
113         spin_unlock_irqrestore(&svc->srv_lock, flags);
114
115         ptlrpc_free_request_buffer (rqbd->rqbd_buffer, svc->srv_buf_size);
116         OBD_FREE (rqbd, sizeof (*rqbd));
117 }
118
119 void
120 ptlrpc_save_lock (struct ptlrpc_request *req, 
121                   struct lustre_handle *lock, int mode)
122 {
123         struct ptlrpc_reply_state *rs = req->rq_reply_state;
124         int                        idx;
125
126         if (!lock->cookie)
127                 return;
128
129         LASSERT (rs != NULL);
130         LASSERT (rs->rs_nlocks < RS_MAX_LOCKS);
131
132         idx = rs->rs_nlocks++;
133         rs->rs_locks[idx] = *lock;
134         rs->rs_modes[idx] = mode;
135         rs->rs_difficult = 1;
136 }
137
138 void
139 ptlrpc_schedule_difficult_reply (struct ptlrpc_reply_state *rs)
140 {
141         struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
142
143 #ifdef CONFIG_SMP
144         LASSERT (spin_is_locked (&svc->srv_lock));
145 #endif
146         LASSERT (rs->rs_difficult);
147         rs->rs_scheduled_ever = 1;              /* flag any notification attempt */
148
149         if (rs->rs_scheduled)                   /* being set up or already notified */
150                 return;
151
152         rs->rs_scheduled = 1;
153         list_del (&rs->rs_list);
154         list_add (&rs->rs_list, &svc->srv_reply_queue);
155         wake_up (&svc->srv_waitq);
156 }
157
158 void 
159 ptlrpc_commit_replies (struct obd_device *obd)
160 {
161         struct list_head   *tmp;
162         struct list_head   *nxt;
163         unsigned long       flags;
164         
165         /* Find any replies that have been committed and get their service
166          * to attend to complete them. */
167
168         /* CAVEAT EMPTOR: spinlock ordering!!! */
169         spin_lock_irqsave (&obd->obd_uncommitted_replies_lock, flags);
170
171         list_for_each_safe (tmp, nxt, &obd->obd_uncommitted_replies) {
172                 struct ptlrpc_reply_state *rs =
173                         list_entry (tmp, struct ptlrpc_reply_state, rs_obd_list);
174
175                 LASSERT (rs->rs_difficult);
176
177                 if (rs->rs_transno <= obd->obd_last_committed) {
178                         struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
179
180                         spin_lock (&svc->srv_lock);
181                         list_del_init (&rs->rs_obd_list);
182                         ptlrpc_schedule_difficult_reply (rs);
183                         spin_unlock (&svc->srv_lock);
184                 }
185         }
186         
187         spin_unlock_irqrestore (&obd->obd_uncommitted_replies_lock, flags);
188 }
189
190 static long
191 timeval_sub(struct timeval *large, struct timeval *small)
192 {
193         return (large->tv_sec - small->tv_sec) * 1000000 +
194                 (large->tv_usec - small->tv_usec);
195 }
196
197 static int
198 ptlrpc_server_post_idle_rqbds (struct ptlrpc_service *svc)
199 {
200         struct ptlrpc_srv_ni              *srv_ni;
201         struct ptlrpc_request_buffer_desc *rqbd;
202         unsigned long                      flags;
203         int                                rc;
204
205         spin_lock_irqsave(&svc->srv_lock, flags);
206         if (list_empty (&svc->srv_idle_rqbds)) {
207                 spin_unlock_irqrestore(&svc->srv_lock, flags);
208                 return (0);
209         }
210
211         rqbd = list_entry(svc->srv_idle_rqbds.next,
212                           struct ptlrpc_request_buffer_desc,
213                           rqbd_list);
214         list_del (&rqbd->rqbd_list);
215
216         /* assume we will post successfully */
217         srv_ni = rqbd->rqbd_srv_ni;
218         srv_ni->sni_nrqbd_receiving++;
219         list_add (&rqbd->rqbd_list, &srv_ni->sni_active_rqbds);
220
221         spin_unlock_irqrestore(&svc->srv_lock, flags);
222
223         rc = ptlrpc_register_rqbd(rqbd);
224         if (rc == 0)
225                 return (1);
226
227         spin_lock_irqsave(&svc->srv_lock, flags);
228
229         srv_ni->sni_nrqbd_receiving--;
230         list_del(&rqbd->rqbd_list);
231         list_add_tail(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
232
233         if (srv_ni->sni_nrqbd_receiving == 0) {
234                 /* This service is off-air on this interface because all
235                  * its request buffers are busy.  Portals will have started
236                  * dropping incoming requests until more buffers get
237                  * posted */
238                 CERROR("All %s %s request buffers busy\n",
239                        svc->srv_name, srv_ni->sni_ni->pni_name);
240         }
241
242         spin_unlock_irqrestore (&svc->srv_lock, flags);
243
244         return (-1);
245 }
246
247 struct ptlrpc_service *
248 ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size,
249                 int req_portal, int rep_portal, 
250                 svc_handler_t handler, char *name,
251                 struct proc_dir_entry *proc_entry)
252 {
253         int                                i;
254         int                                j;
255         int                                ssize;
256         struct ptlrpc_service             *service;
257         struct ptlrpc_srv_ni              *srv_ni;
258         struct ptlrpc_request_buffer_desc *rqbd;
259         ENTRY;
260
261         LASSERT (ptlrpc_ninterfaces > 0);
262         LASSERT (nbufs > 0);
263         LASSERT (bufsize >= max_req_size);
264         
265         ssize = offsetof (struct ptlrpc_service,
266                           srv_interfaces[ptlrpc_ninterfaces]);
267         OBD_ALLOC(service, ssize);
268         if (service == NULL)
269                 RETURN(NULL);
270
271         service->srv_name = name;
272         spin_lock_init(&service->srv_lock);
273         INIT_LIST_HEAD(&service->srv_threads);
274         init_waitqueue_head(&service->srv_waitq);
275
276         service->srv_max_req_size = max_req_size;
277         service->srv_buf_size = bufsize;
278         service->srv_rep_portal = rep_portal;
279         service->srv_req_portal = req_portal;
280         service->srv_handler = handler;
281
282         INIT_LIST_HEAD(&service->srv_request_queue);
283         INIT_LIST_HEAD(&service->srv_idle_rqbds);
284         INIT_LIST_HEAD(&service->srv_reply_queue);
285
286         /* First initialise enough for early teardown */
287         for (i = 0; i < ptlrpc_ninterfaces; i++) {
288                 srv_ni = &service->srv_interfaces[i];
289
290                 srv_ni->sni_service = service;
291                 srv_ni->sni_ni = &ptlrpc_interfaces[i];
292                 INIT_LIST_HEAD(&srv_ni->sni_active_rqbds);
293                 INIT_LIST_HEAD(&srv_ni->sni_active_replies);
294         }
295
296         spin_lock (&ptlrpc_all_services_lock);
297         list_add (&service->srv_list, &ptlrpc_all_services);
298         spin_unlock (&ptlrpc_all_services_lock);
299         
300         /* Now allocate the request buffers, assuming all interfaces require
301          * the same number. */
302         for (i = 0; i < ptlrpc_ninterfaces; i++) {
303                 srv_ni = &service->srv_interfaces[i];
304                 CDEBUG (D_NET, "%s: initialising interface %s\n", name,
305                         srv_ni->sni_ni->pni_name);
306
307                 for (j = 0; j < nbufs; j++) {
308                         rqbd = ptlrpc_alloc_rqbd (srv_ni);
309                         
310                         if (rqbd == NULL) {
311                                 CERROR ("%s.%d: Can't allocate request %d "
312                                         "on %s\n", name, i, j, 
313                                         srv_ni->sni_ni->pni_name);
314                                 GOTO(failed, NULL);
315                         }
316
317                         /* We shouldn't be under memory pressure at
318                          * startup, so fail if we can't post all our
319                          * buffers at this time. */
320                         if (ptlrpc_server_post_idle_rqbds(service) <= 0)
321                                 GOTO(failed, NULL);
322                 }
323         }
324
325         if (proc_entry != NULL)
326                 ptlrpc_lprocfs_register_service(proc_entry, service);
327
328         CDEBUG(D_NET, "%s: Started on %d interfaces, listening on portal %d\n",
329                service->srv_name, ptlrpc_ninterfaces, service->srv_req_portal);
330
331         RETURN(service);
332 failed:
333         ptlrpc_unregister_service(service);
334         return NULL;
335 }
336
337 static void
338 ptlrpc_server_free_request(struct ptlrpc_service *svc, struct ptlrpc_request *req)
339 {
340         unsigned long  flags;
341         int            refcount;
342         
343         spin_lock_irqsave(&svc->srv_lock, flags);
344         svc->srv_n_active_reqs--;
345         refcount = --(req->rq_rqbd->rqbd_refcount);
346         if (refcount == 0) {
347                 /* request buffer is now idle */
348                 list_del(&req->rq_rqbd->rqbd_list);
349                 list_add_tail(&req->rq_rqbd->rqbd_list,
350                               &svc->srv_idle_rqbds);
351         }
352         spin_unlock_irqrestore(&svc->srv_lock, flags);
353
354         ptlrpc_free_server_req(req);
355 }
356
357 static int 
358 ptlrpc_server_handle_request (struct ptlrpc_service *svc)
359 {
360         struct ptlrpc_request *request;
361         unsigned long          flags;
362         struct timeval         work_start;
363         struct timeval         work_end;
364         long                   timediff;
365         int                    rc;
366         char                   str[PTL_NALFMT_SIZE];
367         ENTRY;
368
369         spin_lock_irqsave (&svc->srv_lock, flags);
370         if (list_empty (&svc->srv_request_queue) ||
371             (svc->srv_n_difficult_replies != 0 &&
372              svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
373                 /* If all the other threads are handling requests, I must
374                  * remain free to handle any 'difficult' reply that might
375                  * block them */
376                 spin_unlock_irqrestore (&svc->srv_lock, flags);
377                 RETURN(0);
378         }
379
380         request = list_entry (svc->srv_request_queue.next,
381                               struct ptlrpc_request, rq_list);
382         list_del_init (&request->rq_list);
383         svc->srv_n_queued_reqs--;
384         svc->srv_n_active_reqs++;
385
386         spin_unlock_irqrestore (&svc->srv_lock, flags);
387
388         do_gettimeofday(&work_start);
389         timediff = timeval_sub(&work_start, &request->rq_arrival_time);
390         if (svc->srv_stats != NULL) {
391                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
392                                     timediff);
393                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
394                                     svc->srv_n_queued_reqs);
395                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
396                                     svc->srv_n_active_reqs);
397         }
398
399 #if SWAB_PARANOIA
400         /* Clear request swab mask; this is a new request */
401         request->rq_req_swab_mask = 0;
402 #endif
403         rc = lustre_unpack_msg (request->rq_reqmsg, request->rq_reqlen);
404         if (rc != 0) {
405                 CERROR ("error unpacking request: ptl %d from %s"
406                         " xid "LPU64"\n", svc->srv_req_portal,
407                         ptlrpc_peernid2str(&request->rq_peer, str),
408                        request->rq_xid);
409                 goto out;
410         }
411
412         rc = -EINVAL;
413         if (request->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
414                 CERROR("wrong packet type received (type=%u) from %s\n",
415                        request->rq_reqmsg->type,
416                        ptlrpc_peernid2str(&request->rq_peer, str));
417                 goto out;
418         }
419
420         CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
421
422         /* Discard requests queued for longer than my timeout.  If the
423          * client's timeout is similar to mine, she'll be timing out this
424          * REQ anyway (bug 1502) */
425         if (timediff / 1000000 > (long)obd_timeout) {
426                 CERROR("Dropping timed-out request from %s: %ld seconds old\n",
427                        ptlrpc_peernid2str(&request->rq_peer, str), 
428                        timediff / 1000000);
429                 goto out;
430         }
431
432         request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
433
434         if (request->rq_export) {
435                 if (request->rq_reqmsg->conn_cnt <
436                     request->rq_export->exp_conn_cnt) {
437                         DEBUG_REQ(D_ERROR, request,
438                                   "DROPPING req from old connection %d < %d",
439                                   request->rq_reqmsg->conn_cnt,
440                                   request->rq_export->exp_conn_cnt);
441                         goto put_conn;
442                 }
443
444                 request->rq_export->exp_last_request_time =
445                         LTIME_S(CURRENT_TIME);
446         }
447
448         CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
449                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
450                (request->rq_export ?
451                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
452                (request->rq_export ?
453                 atomic_read(&request->rq_export->exp_refcount) : -99),
454                request->rq_reqmsg->status, request->rq_xid,
455                request->rq_peer.peer_ni->pni_name,
456                ptlrpc_peernid2str(&request->rq_peer, str),
457                request->rq_reqmsg->opc);
458
459         rc = svc->srv_handler(request);
460         CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
461                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
462                (request->rq_export ?
463                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
464                (request->rq_export ?
465                 atomic_read(&request->rq_export->exp_refcount) : -99),
466                request->rq_reqmsg->status, request->rq_xid,
467                request->rq_peer.peer_ni->pni_name,
468                ptlrpc_peernid2str(&request->rq_peer, str),
469                request->rq_reqmsg->opc);
470
471 put_conn:
472         if (request->rq_export != NULL)
473                 class_export_put(request->rq_export);
474
475  out:
476         do_gettimeofday(&work_end);
477
478         timediff = timeval_sub(&work_end, &work_start);
479
480         CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
481                "request "LPU64" opc %u from NID %s processed in %ldus "
482                "(%ldus total)\n", request->rq_xid, request->rq_reqmsg->opc,
483                ptlrpc_peernid2str(&request->rq_peer, str),
484                timediff, timeval_sub(&work_end, &request->rq_arrival_time));
485
486         if (svc->srv_stats != NULL) {
487                 int opc = opcode_offset(request->rq_reqmsg->opc);
488                 if (opc > 0) {
489                         LASSERT(opc < LUSTRE_MAX_OPCODES);
490                         lprocfs_counter_add(svc->srv_stats,
491                                             opc + PTLRPC_LAST_CNTR,
492                                             timediff);
493                 }
494         }
495
496         ptlrpc_server_free_request(svc, request);
497         
498         RETURN(1);
499 }
500
501 static int
502 ptlrpc_server_handle_reply (struct ptlrpc_service *svc) 
503 {
504         struct ptlrpc_reply_state *rs;
505         unsigned long              flags;
506         struct obd_export         *exp;
507         struct obd_device         *obd;
508         int                        nlocks;
509         int                        been_handled;
510         char                       str[PTL_NALFMT_SIZE];
511         ENTRY;
512
513         spin_lock_irqsave (&svc->srv_lock, flags);
514         if (list_empty (&svc->srv_reply_queue)) {
515                 spin_unlock_irqrestore (&svc->srv_lock, flags);
516                 RETURN(0);
517         }
518         
519         rs = list_entry (svc->srv_reply_queue.next,
520                          struct ptlrpc_reply_state, rs_list);
521
522         exp = rs->rs_export;
523         obd = exp->exp_obd;
524
525         LASSERT (rs->rs_difficult);
526         LASSERT (rs->rs_scheduled);
527
528         list_del_init (&rs->rs_list);
529
530         /* Disengage from notifiers carefully (lock ordering!) */
531         spin_unlock(&svc->srv_lock);
532
533         spin_lock (&obd->obd_uncommitted_replies_lock);
534         /* Noop if removed already */
535         list_del_init (&rs->rs_obd_list);
536         spin_unlock (&obd->obd_uncommitted_replies_lock);
537
538         spin_lock (&exp->exp_lock);
539         /* Noop if removed already */
540         list_del_init (&rs->rs_exp_list);
541         spin_unlock (&exp->exp_lock);
542
543         spin_lock(&svc->srv_lock);
544
545         been_handled = rs->rs_handled;
546         rs->rs_handled = 1;
547         
548         nlocks = rs->rs_nlocks;                 /* atomic "steal", but */
549         rs->rs_nlocks = 0;                      /* locks still on rs_locks! */
550
551         if (nlocks == 0 && !been_handled) {
552                 /* If we see this, we should already have seen the warning
553                  * in mds_steal_ack_locks()  */
554                 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
555                       " o%d NID %s\n",
556                       rs, 
557                       rs->rs_xid, rs->rs_transno,
558                       rs->rs_msg.opc, 
559                       ptlrpc_peernid2str(&exp->exp_connection->c_peer, str));
560         }
561
562         if ((!been_handled && rs->rs_on_net) || 
563             nlocks > 0) {
564                 spin_unlock_irqrestore(&svc->srv_lock, flags);
565                 
566                 if (!been_handled && rs->rs_on_net) {
567                         PtlMDUnlink(rs->rs_md_h);
568                         /* Ignore return code; we're racing with
569                          * completion... */
570                 }
571
572                 while (nlocks-- > 0)
573                         ldlm_lock_decref(&rs->rs_locks[nlocks], 
574                                          rs->rs_modes[nlocks]);
575
576                 spin_lock_irqsave(&svc->srv_lock, flags);
577         }
578
579         rs->rs_scheduled = 0;
580
581         if (!rs->rs_on_net) {
582                 /* Off the net */
583                 svc->srv_n_difficult_replies--;
584                 spin_unlock_irqrestore(&svc->srv_lock, flags);
585                 
586                 class_export_put (exp);
587                 rs->rs_export = NULL;
588                 lustre_free_reply_state (rs);
589                 atomic_dec (&svc->srv_outstanding_replies);
590                 RETURN(1);
591         }
592         
593         /* still on the net; callback will schedule */
594         spin_unlock_irqrestore (&svc->srv_lock, flags);
595         RETURN(1);
596 }
597
598 #ifndef __KERNEL__
599 /* FIXME make use of timeout later */
600 int
601 liblustre_check_services (void *arg) 
602 {
603         int  did_something = 0;
604         int  rc;
605         struct list_head *tmp, *nxt;
606         ENTRY;
607         
608         /* I'm relying on being single threaded, not to have to lock
609          * ptlrpc_all_services etc */
610         list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
611                 struct ptlrpc_service *svc =
612                         list_entry (tmp, struct ptlrpc_service, srv_list);
613                 
614                 if (svc->srv_nthreads != 0)     /* I've recursed */
615                         continue;
616
617                 /* service threads can block for bulk, so this limits us
618                  * (arbitrarily) to recursing 1 stack frame per service.
619                  * Note that the problem with recursion is that we have to
620                  * unwind completely before our caller can resume. */
621                 
622                 svc->srv_nthreads++;
623                 
624                 do {
625                         rc = ptlrpc_server_handle_reply(svc);
626                         rc |= ptlrpc_server_handle_request(svc);
627                         rc |= (ptlrpc_server_post_idle_rqbds(svc) > 0);
628                         did_something |= rc;
629                 } while (rc);
630                 
631                 svc->srv_nthreads--;
632         }
633
634         RETURN(did_something);
635 }
636
637 #else /* __KERNEL__ */
638
639 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
640 void ptlrpc_daemonize(void)
641 {
642         exit_mm(current);
643         lustre_daemonize_helper();
644         exit_files(current);
645         reparent_to_init();
646 }
647
648 static int
649 ptlrpc_retry_rqbds(void *arg)
650 {
651         struct ptlrpc_service *svc = (struct ptlrpc_service *)arg;
652         
653         svc->srv_rqbd_timeout = 0;
654         return (-ETIMEDOUT);
655 }
656
657 static int ptlrpc_main(void *arg)
658 {
659         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
660         struct ptlrpc_service  *svc = data->svc;
661         struct ptlrpc_thread   *thread = data->thread;
662         unsigned long           flags;
663         ENTRY;
664
665         lock_kernel();
666         ptlrpc_daemonize();
667
668         SIGNAL_MASK_LOCK(current, flags);
669         sigfillset(&current->blocked);
670         RECALC_SIGPENDING;
671         SIGNAL_MASK_UNLOCK(current, flags);
672
673         LASSERTF(strlen(data->name) < sizeof(current->comm),
674                  "name %d > len %d\n",
675                  (int)strlen(data->name), (int)sizeof(current->comm));
676         THREAD_NAME(current->comm, sizeof(current->comm) - 1, "%s", data->name);
677         
678         unlock_kernel();
679
680         /* Record that the thread is running */
681         thread->t_flags = SVC_RUNNING;
682         wake_up(&thread->t_ctl_waitq);
683
684         spin_lock_irqsave(&svc->srv_lock, flags);
685         svc->srv_nthreads++;
686         spin_unlock_irqrestore(&svc->srv_lock, flags);
687
688         /* XXX maintain a list of all managed devices: insert here */
689
690         while ((thread->t_flags & SVC_STOPPING) == 0 ||
691                svc->srv_n_difficult_replies != 0) {
692                 /* Don't exit while there are replies to be handled */
693                 struct l_wait_info lwi = LWI_TIMEOUT(svc->srv_rqbd_timeout,
694                                                      ptlrpc_retry_rqbds, svc);
695
696                 l_wait_event_exclusive (svc->srv_waitq,
697                               ((thread->t_flags & SVC_STOPPING) != 0 &&
698                                svc->srv_n_difficult_replies == 0) ||
699                               (!list_empty(&svc->srv_idle_rqbds) &&
700                                svc->srv_rqbd_timeout == 0) ||
701                               !list_empty (&svc->srv_reply_queue) ||
702                               (!list_empty (&svc->srv_request_queue) &&
703                                (svc->srv_n_difficult_replies == 0 ||
704                                 svc->srv_n_active_reqs <
705                                 (svc->srv_nthreads - 1))),
706                               &lwi);
707
708                 if (!list_empty (&svc->srv_reply_queue))
709                         ptlrpc_server_handle_reply (svc);
710
711                 /* only handle requests if there are no difficult replies
712                  * outstanding, or I'm not the last thread handling
713                  * requests */
714                 if (!list_empty (&svc->srv_request_queue) &&
715                     (svc->srv_n_difficult_replies == 0 ||
716                      svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
717                         ptlrpc_server_handle_request (svc);
718
719                 if (!list_empty(&svc->srv_idle_rqbds) &&
720                     ptlrpc_server_post_idle_rqbds(svc) < 0) {
721                         /* I just failed to repost request buffers.  Wait
722                          * for a timeout (unless something else happens)
723                          * before I try again */
724                         svc->srv_rqbd_timeout = HZ/10;
725                 }
726         }
727
728         spin_lock_irqsave(&svc->srv_lock, flags);
729
730         svc->srv_nthreads--;                    /* must know immediately */
731         thread->t_flags = SVC_STOPPED;
732         wake_up(&thread->t_ctl_waitq);
733
734         spin_unlock_irqrestore(&svc->srv_lock, flags);
735
736         CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
737         return 0;
738 }
739
740 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
741                                struct ptlrpc_thread *thread)
742 {
743         struct l_wait_info lwi = { 0 };
744         unsigned long      flags;
745
746         spin_lock_irqsave(&svc->srv_lock, flags);
747         thread->t_flags = SVC_STOPPING;
748         spin_unlock_irqrestore(&svc->srv_lock, flags);
749
750         wake_up_all(&svc->srv_waitq);
751         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
752                      &lwi);
753
754         spin_lock_irqsave(&svc->srv_lock, flags);
755         list_del(&thread->t_link);
756         spin_unlock_irqrestore(&svc->srv_lock, flags);
757
758         OBD_FREE(thread, sizeof(*thread));
759 }
760
761 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
762 {
763         unsigned long flags;
764         struct ptlrpc_thread *thread;
765
766         spin_lock_irqsave(&svc->srv_lock, flags);
767         while (!list_empty(&svc->srv_threads)) {
768                 thread = list_entry(svc->srv_threads.next,
769                                     struct ptlrpc_thread, t_link);
770
771                 spin_unlock_irqrestore(&svc->srv_lock, flags);
772                 ptlrpc_stop_thread(svc, thread);
773                 spin_lock_irqsave(&svc->srv_lock, flags);
774         }
775
776         spin_unlock_irqrestore(&svc->srv_lock, flags);
777 }
778
779 /* @base_name should be 12 characters or less - 3 will be added on */
780 int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
781                            int num_threads, char *base_name)
782 {
783         int i, rc = 0;
784         ENTRY;
785
786         for (i = 0; i < num_threads; i++) {
787                 char name[32];
788                 sprintf(name, "%s_%02d", base_name, i);
789                 rc = ptlrpc_start_thread(dev, svc, name);
790                 if (rc) {
791                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
792                                i, rc);
793                         ptlrpc_stop_all_threads(svc);
794                 }
795         }
796         RETURN(rc);
797 }
798
799 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
800                         char *name)
801 {
802         struct l_wait_info lwi = { 0 };
803         struct ptlrpc_svc_data d;
804         struct ptlrpc_thread *thread;
805         unsigned long flags;
806         int rc;
807         ENTRY;
808
809         OBD_ALLOC(thread, sizeof(*thread));
810         if (thread == NULL)
811                 RETURN(-ENOMEM);
812         init_waitqueue_head(&thread->t_ctl_waitq);
813         
814         d.dev = dev;
815         d.svc = svc;
816         d.name = name;
817         d.thread = thread;
818
819         spin_lock_irqsave(&svc->srv_lock, flags);
820         list_add(&thread->t_link, &svc->srv_threads);
821         spin_unlock_irqrestore(&svc->srv_lock, flags);
822
823         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
824          * just drop the VM and FILES in ptlrpc_daemonize() right away.
825          */
826         rc = kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
827         if (rc < 0) {
828                 CERROR("cannot start thread: %d\n", rc);
829                 OBD_FREE(thread, sizeof(*thread));
830                 RETURN(rc);
831         }
832         l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
833
834         RETURN(0);
835 }
836 #endif
837
838 int ptlrpc_unregister_service(struct ptlrpc_service *service)
839 {
840         int                   i;
841         int                   rc;
842         unsigned long         flags;
843         struct ptlrpc_srv_ni *srv_ni;
844         struct l_wait_info    lwi;
845         struct list_head     *tmp;
846
847         LASSERT(list_empty(&service->srv_threads));
848
849         spin_lock (&ptlrpc_all_services_lock);
850         list_del_init (&service->srv_list);
851         spin_unlock (&ptlrpc_all_services_lock);
852
853         ptlrpc_lprocfs_unregister_service(service);
854
855         for (i = 0; i < ptlrpc_ninterfaces; i++) {
856                 srv_ni = &service->srv_interfaces[i];
857                 CDEBUG(D_NET, "%s: tearing down interface %s\n",
858                        service->srv_name, srv_ni->sni_ni->pni_name);
859
860                 /* Unlink all the request buffers.  This forces a 'final'
861                  * event with its 'unlink' flag set for each posted rqbd */
862                 list_for_each(tmp, &srv_ni->sni_active_rqbds) {
863                         struct ptlrpc_request_buffer_desc *rqbd =
864                                 list_entry(tmp, struct ptlrpc_request_buffer_desc, 
865                                            rqbd_list);
866
867                         rc = PtlMDUnlink(rqbd->rqbd_md_h);
868                         LASSERT (rc == PTL_OK || rc == PTL_MD_INVALID);
869                 }
870
871                 /* Wait for the network to release any buffers it's
872                  * currently filling */
873                 for (;;) {
874                         spin_lock_irqsave(&service->srv_lock, flags);
875                         rc = srv_ni->sni_nrqbd_receiving;
876                         spin_unlock_irqrestore(&service->srv_lock, flags);
877
878                         if (rc == 0)
879                                 break;
880                         
881                         /* Network access will complete in finite time but
882                          * the HUGE timeout lets us CWARN for visibility of
883                          * sluggish NALs */
884                         lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
885                         rc = l_wait_event(service->srv_waitq,
886                                           srv_ni->sni_nrqbd_receiving == 0,
887                                           &lwi);
888                         if (rc == -ETIMEDOUT)
889                                 CWARN("Waiting for request buffers on "
890                                       "service %s on interface %s ",
891                                       service->srv_name, srv_ni->sni_ni->pni_name);
892                 }
893
894                 /* schedule all outstanding replies to terminate them */
895                 spin_lock_irqsave(&service->srv_lock, flags);
896                 while (!list_empty(&srv_ni->sni_active_replies)) {
897                         struct ptlrpc_reply_state *rs =
898                                 list_entry(srv_ni->sni_active_replies.next,
899                                            struct ptlrpc_reply_state,
900                                            rs_list);
901                         ptlrpc_schedule_difficult_reply(rs);
902                 }
903                 spin_unlock_irqrestore(&service->srv_lock, flags);
904         }
905
906         /* purge the request queue.  NB No new replies (rqbds all unlinked)
907          * and no service threads, so I'm the only thread noodling the
908          * request queue now */
909         while (!list_empty(&service->srv_request_queue)) {
910                 struct ptlrpc_request *req =
911                         list_entry(service->srv_request_queue.next,
912                                    struct ptlrpc_request,
913                                    rq_list);
914                 
915                 list_del(&req->rq_list);
916                 service->srv_n_queued_reqs--;
917                 service->srv_n_active_reqs++;
918
919                 ptlrpc_server_free_request(service, req);
920         }
921         LASSERT(service->srv_n_queued_reqs == 0);
922         LASSERT(service->srv_n_active_reqs == 0);
923
924         for (i = 0; i < ptlrpc_ninterfaces; i++) {
925                 srv_ni = &service->srv_interfaces[i];
926                 LASSERT(list_empty(&srv_ni->sni_active_rqbds));
927         }
928
929         /* Now free all the request buffers since nothing references them
930          * any more... */
931         while (!list_empty(&service->srv_idle_rqbds)) {
932                 struct ptlrpc_request_buffer_desc *rqbd =
933                         list_entry(service->srv_idle_rqbds.next,
934                                    struct ptlrpc_request_buffer_desc, 
935                                    rqbd_list);
936
937                 ptlrpc_free_rqbd(rqbd);
938         }
939
940         /* wait for all outstanding replies to complete (they were
941          * scheduled having been flagged to abort above) */
942         while (atomic_read(&service->srv_outstanding_replies) != 0) {
943                 struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
944
945                 rc = l_wait_event(service->srv_waitq,
946                                   !list_empty(&service->srv_reply_queue), &lwi);
947                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
948
949                 if (rc == 0) {
950                         ptlrpc_server_handle_reply(service);
951                         continue;
952                 }
953                 CWARN("Unexpectedly long timeout %p\n", service);
954         }
955
956         OBD_FREE(service,
957                  offsetof(struct ptlrpc_service,
958                           srv_interfaces[ptlrpc_ninterfaces]));
959         return 0;
960 }