1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2002 Cluster File Systems, Inc.
6 * This file is part of Lustre, http://www.lustre.org.
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.
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.
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.
23 #define DEBUG_SUBSYSTEM S_RPC
25 #include <liblustre.h>
26 #include <linux/kp30.h>
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"
34 static LIST_HEAD (ptlrpc_all_services);
35 static spinlock_t ptlrpc_all_services_lock = SPIN_LOCK_UNLOCKED;
38 ptlrpc_free_server_req (struct ptlrpc_request *req)
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)
46 OBD_FREE(req, sizeof(*req));
50 ptlrpc_alloc_request_buffer (int size)
54 if (size > SVC_BUF_VMALLOC_THRESHOLD)
55 OBD_VMALLOC(ptr, size);
63 ptlrpc_free_request_buffer (char *ptr, int size)
65 if (size > SVC_BUF_VMALLOC_THRESHOLD)
71 struct ptlrpc_request_buffer_desc *
72 ptlrpc_alloc_rqbd (struct ptlrpc_srv_ni *srv_ni)
74 struct ptlrpc_service *svc = srv_ni->sni_service;
76 struct ptlrpc_request_buffer_desc *rqbd;
78 OBD_ALLOC(rqbd, sizeof (*rqbd));
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);
88 if (rqbd->rqbd_buffer == NULL) {
89 OBD_FREE(rqbd, sizeof (*rqbd));
93 spin_lock_irqsave (&svc->srv_lock, flags);
94 list_add(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
96 spin_unlock_irqrestore (&svc->srv_lock, flags);
102 ptlrpc_free_rqbd (struct ptlrpc_request_buffer_desc *rqbd)
104 struct ptlrpc_srv_ni *sni = rqbd->rqbd_srv_ni;
105 struct ptlrpc_service *svc = sni->sni_service;
108 LASSERT (rqbd->rqbd_refcount == 0);
110 spin_lock_irqsave(&svc->srv_lock, flags);
111 list_del(&rqbd->rqbd_list);
113 spin_unlock_irqrestore(&svc->srv_lock, flags);
115 ptlrpc_free_request_buffer (rqbd->rqbd_buffer, svc->srv_buf_size);
116 OBD_FREE (rqbd, sizeof (*rqbd));
120 ptlrpc_save_lock (struct ptlrpc_request *req,
121 struct lustre_handle *lock, int mode)
123 struct ptlrpc_reply_state *rs = req->rq_reply_state;
126 LASSERT (rs != NULL);
127 LASSERT (rs->rs_nlocks < RS_MAX_LOCKS);
129 idx = rs->rs_nlocks++;
130 rs->rs_locks[idx] = *lock;
131 rs->rs_modes[idx] = mode;
132 rs->rs_difficult = 1;
136 ptlrpc_schedule_difficult_reply (struct ptlrpc_reply_state *rs)
138 struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
141 LASSERT (spin_is_locked (&svc->srv_lock));
143 LASSERT (rs->rs_difficult);
144 rs->rs_scheduled_ever = 1; /* flag any notification attempt */
146 if (rs->rs_scheduled) /* being set up or already notified */
149 rs->rs_scheduled = 1;
150 list_del (&rs->rs_list);
151 list_add (&rs->rs_list, &svc->srv_reply_queue);
152 wake_up (&svc->srv_waitq);
156 ptlrpc_commit_replies (struct obd_device *obd)
158 struct list_head *tmp;
159 struct list_head *nxt;
162 /* Find any replies that have been committed and get their service
163 * to attend to complete them. */
165 /* CAVEAT EMPTOR: spinlock ordering!!! */
166 spin_lock_irqsave (&obd->obd_uncommitted_replies_lock, flags);
168 list_for_each_safe (tmp, nxt, &obd->obd_uncommitted_replies) {
169 struct ptlrpc_reply_state *rs =
170 list_entry (tmp, struct ptlrpc_reply_state, rs_obd_list);
172 LASSERT (rs->rs_difficult);
174 if (rs->rs_transno <= obd->obd_last_committed) {
175 struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
177 spin_lock (&svc->srv_lock);
178 list_del_init (&rs->rs_obd_list);
179 ptlrpc_schedule_difficult_reply (rs);
180 spin_unlock (&svc->srv_lock);
184 spin_unlock_irqrestore (&obd->obd_uncommitted_replies_lock, flags);
188 timeval_sub(struct timeval *large, struct timeval *small)
190 return (large->tv_sec - small->tv_sec) * 1000000 +
191 (large->tv_usec - small->tv_usec);
195 ptlrpc_server_post_idle_rqbds (struct ptlrpc_service *svc)
197 struct ptlrpc_srv_ni *srv_ni;
198 struct ptlrpc_request_buffer_desc *rqbd;
202 spin_lock_irqsave(&svc->srv_lock, flags);
203 if (list_empty (&svc->srv_idle_rqbds)) {
204 spin_unlock_irqrestore(&svc->srv_lock, flags);
208 rqbd = list_entry(svc->srv_idle_rqbds.next,
209 struct ptlrpc_request_buffer_desc,
211 list_del (&rqbd->rqbd_list);
213 /* assume we will post successfully */
214 srv_ni = rqbd->rqbd_srv_ni;
215 srv_ni->sni_nrqbd_receiving++;
216 list_add (&rqbd->rqbd_list, &srv_ni->sni_active_rqbds);
218 spin_unlock_irqrestore(&svc->srv_lock, flags);
220 rc = ptlrpc_register_rqbd(rqbd);
224 spin_lock_irqsave(&svc->srv_lock, flags);
226 srv_ni->sni_nrqbd_receiving--;
227 list_del(&rqbd->rqbd_list);
228 list_add_tail(&rqbd->rqbd_list, &svc->srv_idle_rqbds);
230 if (srv_ni->sni_nrqbd_receiving == 0) {
231 /* This service is off-air on this interface because all
232 * its request buffers are busy. Portals will have started
233 * dropping incoming requests until more buffers get
235 CERROR("All %s %s request buffers busy\n",
236 svc->srv_name, srv_ni->sni_ni->pni_name);
239 spin_unlock_irqrestore (&svc->srv_lock, flags);
244 struct ptlrpc_service *
245 ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size,
246 int req_portal, int rep_portal,
247 svc_handler_t handler, char *name,
248 struct proc_dir_entry *proc_entry)
253 struct ptlrpc_service *service;
254 struct ptlrpc_srv_ni *srv_ni;
255 struct ptlrpc_request_buffer_desc *rqbd;
258 LASSERT (ptlrpc_ninterfaces > 0);
260 LASSERT (bufsize >= max_req_size);
262 ssize = offsetof (struct ptlrpc_service,
263 srv_interfaces[ptlrpc_ninterfaces]);
264 OBD_ALLOC(service, ssize);
268 service->srv_name = name;
269 spin_lock_init(&service->srv_lock);
270 INIT_LIST_HEAD(&service->srv_threads);
271 init_waitqueue_head(&service->srv_waitq);
273 service->srv_max_req_size = max_req_size;
274 service->srv_buf_size = bufsize;
275 service->srv_rep_portal = rep_portal;
276 service->srv_req_portal = req_portal;
277 service->srv_handler = handler;
279 INIT_LIST_HEAD(&service->srv_request_queue);
280 INIT_LIST_HEAD(&service->srv_idle_rqbds);
281 INIT_LIST_HEAD(&service->srv_reply_queue);
283 /* First initialise enough for early teardown */
284 for (i = 0; i < ptlrpc_ninterfaces; i++) {
285 srv_ni = &service->srv_interfaces[i];
287 srv_ni->sni_service = service;
288 srv_ni->sni_ni = &ptlrpc_interfaces[i];
289 INIT_LIST_HEAD(&srv_ni->sni_active_rqbds);
290 INIT_LIST_HEAD(&srv_ni->sni_active_replies);
293 spin_lock (&ptlrpc_all_services_lock);
294 list_add (&service->srv_list, &ptlrpc_all_services);
295 spin_unlock (&ptlrpc_all_services_lock);
297 /* Now allocate the request buffers, assuming all interfaces require
298 * the same number. */
299 for (i = 0; i < ptlrpc_ninterfaces; i++) {
300 srv_ni = &service->srv_interfaces[i];
301 CDEBUG (D_NET, "%s: initialising interface %s\n", name,
302 srv_ni->sni_ni->pni_name);
304 for (j = 0; j < nbufs; j++) {
305 rqbd = ptlrpc_alloc_rqbd (srv_ni);
308 CERROR ("%s.%d: Can't allocate request %d "
309 "on %s\n", name, i, j,
310 srv_ni->sni_ni->pni_name);
314 /* We shouldn't be under memory pressure at
315 * startup, so fail if we can't post all our
316 * buffers at this time. */
317 if (ptlrpc_server_post_idle_rqbds(service) <= 0)
322 if (proc_entry != NULL)
323 ptlrpc_lprocfs_register_service(proc_entry, service);
325 CDEBUG(D_NET, "%s: Started on %d interfaces, listening on portal %d\n",
326 service->srv_name, ptlrpc_ninterfaces, service->srv_req_portal);
330 ptlrpc_unregister_service(service);
335 ptlrpc_server_free_request(struct ptlrpc_service *svc, struct ptlrpc_request *req)
340 spin_lock_irqsave(&svc->srv_lock, flags);
341 svc->srv_n_active_reqs--;
342 refcount = --(req->rq_rqbd->rqbd_refcount);
344 /* request buffer is now idle */
345 list_del(&req->rq_rqbd->rqbd_list);
346 list_add_tail(&req->rq_rqbd->rqbd_list,
347 &svc->srv_idle_rqbds);
349 spin_unlock_irqrestore(&svc->srv_lock, flags);
351 ptlrpc_free_server_req(req);
355 ptlrpc_server_handle_request (struct ptlrpc_service *svc)
357 struct ptlrpc_request *request;
359 struct timeval work_start;
360 struct timeval work_end;
365 spin_lock_irqsave (&svc->srv_lock, flags);
366 if (list_empty (&svc->srv_request_queue) ||
367 (svc->srv_n_difficult_replies != 0 &&
368 svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
369 /* If all the other threads are handling requests, I must
370 * remain free to handle any 'difficult' reply that might
372 spin_unlock_irqrestore (&svc->srv_lock, flags);
376 request = list_entry (svc->srv_request_queue.next,
377 struct ptlrpc_request, rq_list);
378 list_del_init (&request->rq_list);
379 svc->srv_n_queued_reqs--;
380 svc->srv_n_active_reqs++;
382 spin_unlock_irqrestore (&svc->srv_lock, flags);
384 do_gettimeofday(&work_start);
385 timediff = timeval_sub(&work_start, &request->rq_arrival_time);
386 if (svc->srv_stats != NULL) {
387 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
389 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
390 svc->srv_n_queued_reqs);
391 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
392 svc->srv_n_active_reqs);
396 /* Clear request swab mask; this is a new request */
397 request->rq_req_swab_mask = 0;
399 rc = lustre_unpack_msg (request->rq_reqmsg, request->rq_reqlen);
401 CERROR ("error unpacking request: ptl %d from "LPX64
402 " xid "LPU64"\n", svc->srv_req_portal,
403 request->rq_peer.peer_nid, request->rq_xid);
408 if (request->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
409 CERROR("wrong packet type received (type=%u) from "
410 LPX64"\n", request->rq_reqmsg->type,
411 request->rq_peer.peer_nid);
415 CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
417 /* Discard requests queued for longer than my timeout. If the
418 * client's timeout is similar to mine, she'll be timing out this
419 * REQ anyway (bug 1502) */
420 if (timediff / 1000000 > (long)obd_timeout) {
421 CERROR("Dropping timed-out request from "LPX64
422 ": %ld seconds old\n",
423 request->rq_peer.peer_nid, timediff / 1000000);
427 request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
429 if (request->rq_export) {
430 if (request->rq_reqmsg->conn_cnt <
431 request->rq_export->exp_conn_cnt) {
432 DEBUG_REQ(D_ERROR, request,
433 "DROPPING req from old connection %d < %d",
434 request->rq_reqmsg->conn_cnt,
435 request->rq_export->exp_conn_cnt);
439 request->rq_export->exp_last_request_time =
440 LTIME_S(CURRENT_TIME);
443 CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
444 "%s:%s+%d:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
445 (request->rq_export ?
446 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
447 (request->rq_export ?
448 atomic_read(&request->rq_export->exp_refcount) : -99),
449 request->rq_reqmsg->status, request->rq_xid,
450 request->rq_peer.peer_ni->pni_name,
451 request->rq_peer.peer_nid,
452 request->rq_reqmsg->opc);
454 rc = svc->srv_handler(request);
455 CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
456 "%s:%s+%d:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
457 (request->rq_export ?
458 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
459 (request->rq_export ?
460 atomic_read(&request->rq_export->exp_refcount) : -99),
461 request->rq_reqmsg->status, request->rq_xid,
462 request->rq_peer.peer_ni->pni_name,
463 request->rq_peer.peer_nid,
464 request->rq_reqmsg->opc);
467 if (request->rq_export != NULL)
468 class_export_put(request->rq_export);
471 do_gettimeofday(&work_end);
473 timediff = timeval_sub(&work_end, &work_start);
475 CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
476 "request "LPU64" opc %u from NID "LPX64" processed in %ldus "
477 "(%ldus total)\n", request->rq_xid, request->rq_reqmsg->opc,
478 request->rq_peer.peer_nid,
479 timediff, timeval_sub(&work_end, &request->rq_arrival_time));
481 if (svc->srv_stats != NULL) {
482 int opc = opcode_offset(request->rq_reqmsg->opc);
484 LASSERT(opc < LUSTRE_MAX_OPCODES);
485 lprocfs_counter_add(svc->srv_stats,
486 opc + PTLRPC_LAST_CNTR,
491 ptlrpc_server_free_request(svc, request);
497 ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
499 struct ptlrpc_reply_state *rs;
501 struct obd_export *exp;
502 struct obd_device *obd;
507 spin_lock_irqsave (&svc->srv_lock, flags);
508 if (list_empty (&svc->srv_reply_queue)) {
509 spin_unlock_irqrestore (&svc->srv_lock, flags);
513 rs = list_entry (svc->srv_reply_queue.next,
514 struct ptlrpc_reply_state, rs_list);
519 LASSERT (rs->rs_difficult);
520 LASSERT (rs->rs_scheduled);
522 list_del_init (&rs->rs_list);
524 /* Disengage from notifiers carefully (lock ordering!) */
525 spin_unlock(&svc->srv_lock);
527 spin_lock (&obd->obd_uncommitted_replies_lock);
528 /* Noop if removed already */
529 list_del_init (&rs->rs_obd_list);
530 spin_unlock (&obd->obd_uncommitted_replies_lock);
532 spin_lock (&exp->exp_lock);
533 /* Noop if removed already */
534 list_del_init (&rs->rs_exp_list);
535 spin_unlock (&exp->exp_lock);
537 spin_lock(&svc->srv_lock);
539 been_handled = rs->rs_handled;
542 nlocks = rs->rs_nlocks; /* atomic "steal", but */
543 rs->rs_nlocks = 0; /* locks still on rs_locks! */
545 if (nlocks == 0 && !been_handled) {
546 /* If we see this, we should already have seen the warning
547 * in mds_steal_ack_locks() */
548 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
551 rs->rs_xid, rs->rs_transno,
552 rs->rs_msg.opc, exp->exp_connection->c_peer.peer_nid);
555 if ((!been_handled && rs->rs_on_net) ||
557 spin_unlock_irqrestore(&svc->srv_lock, flags);
559 if (!been_handled && rs->rs_on_net) {
560 PtlMDUnlink(rs->rs_md_h);
561 /* Ignore return code; we're racing with
566 ldlm_lock_decref(&rs->rs_locks[nlocks],
567 rs->rs_modes[nlocks]);
569 spin_lock_irqsave(&svc->srv_lock, flags);
572 rs->rs_scheduled = 0;
574 if (!rs->rs_on_net) {
576 svc->srv_n_difficult_replies--;
577 spin_unlock_irqrestore(&svc->srv_lock, flags);
579 class_export_put (exp);
580 rs->rs_export = NULL;
581 lustre_free_reply_state (rs);
582 atomic_dec (&svc->srv_outstanding_replies);
586 /* still on the net; callback will schedule */
587 spin_unlock_irqrestore (&svc->srv_lock, flags);
592 /* FIXME make use of timeout later */
594 liblustre_check_services (void *arg)
596 int did_something = 0;
598 struct list_head *tmp, *nxt;
601 /* I'm relying on being single threaded, not to have to lock
602 * ptlrpc_all_services etc */
603 list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
604 struct ptlrpc_service *svc =
605 list_entry (tmp, struct ptlrpc_service, srv_list);
607 if (svc->srv_nthreads != 0) /* I've recursed */
610 /* service threads can block for bulk, so this limits us
611 * (arbitrarily) to recursing 1 stack frame per service.
612 * Note that the problem with recursion is that we have to
613 * unwind completely before our caller can resume. */
618 rc = ptlrpc_server_handle_reply(svc);
619 rc |= ptlrpc_server_handle_request(svc);
620 rc |= (ptlrpc_server_post_idle_rqbds(svc) > 0);
627 RETURN(did_something);
630 #else /* __KERNEL__ */
632 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
633 void ptlrpc_daemonize(void)
636 lustre_daemonize_helper();
642 ptlrpc_retry_rqbds(void *arg)
644 struct ptlrpc_service *svc = (struct ptlrpc_service *)arg;
646 svc->srv_rqbd_timeout = 0;
650 static int ptlrpc_main(void *arg)
652 struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
653 struct ptlrpc_service *svc = data->svc;
654 struct ptlrpc_thread *thread = data->thread;
661 SIGNAL_MASK_LOCK(current, flags);
662 sigfillset(¤t->blocked);
664 SIGNAL_MASK_UNLOCK(current, flags);
666 THREAD_NAME(current->comm, "%s", data->name);
669 /* Record that the thread is running */
670 thread->t_flags = SVC_RUNNING;
671 wake_up(&thread->t_ctl_waitq);
673 spin_lock_irqsave(&svc->srv_lock, flags);
675 spin_unlock_irqrestore(&svc->srv_lock, flags);
677 /* XXX maintain a list of all managed devices: insert here */
679 while ((thread->t_flags & SVC_STOPPING) == 0 ||
680 svc->srv_n_difficult_replies != 0) {
681 /* Don't exit while there are replies to be handled */
682 struct l_wait_info lwi = LWI_TIMEOUT(svc->srv_rqbd_timeout,
683 ptlrpc_retry_rqbds, svc);
685 l_wait_event_exclusive (svc->srv_waitq,
686 ((thread->t_flags & SVC_STOPPING) != 0 &&
687 svc->srv_n_difficult_replies == 0) ||
688 (!list_empty(&svc->srv_idle_rqbds) &&
689 svc->srv_rqbd_timeout == 0) ||
690 !list_empty (&svc->srv_reply_queue) ||
691 (!list_empty (&svc->srv_request_queue) &&
692 (svc->srv_n_difficult_replies == 0 ||
693 svc->srv_n_active_reqs <
694 (svc->srv_nthreads - 1))),
697 if (!list_empty (&svc->srv_reply_queue))
698 ptlrpc_server_handle_reply (svc);
700 /* only handle requests if there are no difficult replies
701 * outstanding, or I'm not the last thread handling
703 if (!list_empty (&svc->srv_request_queue) &&
704 (svc->srv_n_difficult_replies == 0 ||
705 svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
706 ptlrpc_server_handle_request (svc);
708 if (!list_empty(&svc->srv_idle_rqbds) &&
709 ptlrpc_server_post_idle_rqbds(svc) < 0) {
710 /* I just failed to repost request buffers. Wait
711 * for a timeout (unless something else happens)
712 * before I try again */
713 svc->srv_rqbd_timeout = HZ/10;
717 spin_lock_irqsave(&svc->srv_lock, flags);
719 svc->srv_nthreads--; /* must know immediately */
720 thread->t_flags = SVC_STOPPED;
721 wake_up(&thread->t_ctl_waitq);
723 spin_unlock_irqrestore(&svc->srv_lock, flags);
725 CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
729 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
730 struct ptlrpc_thread *thread)
732 struct l_wait_info lwi = { 0 };
735 spin_lock_irqsave(&svc->srv_lock, flags);
736 thread->t_flags = SVC_STOPPING;
737 spin_unlock_irqrestore(&svc->srv_lock, flags);
739 wake_up_all(&svc->srv_waitq);
740 l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
743 spin_lock_irqsave(&svc->srv_lock, flags);
744 list_del(&thread->t_link);
745 spin_unlock_irqrestore(&svc->srv_lock, flags);
747 OBD_FREE(thread, sizeof(*thread));
750 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
753 struct ptlrpc_thread *thread;
755 spin_lock_irqsave(&svc->srv_lock, flags);
756 while (!list_empty(&svc->srv_threads)) {
757 thread = list_entry(svc->srv_threads.next,
758 struct ptlrpc_thread, t_link);
760 spin_unlock_irqrestore(&svc->srv_lock, flags);
761 ptlrpc_stop_thread(svc, thread);
762 spin_lock_irqsave(&svc->srv_lock, flags);
765 spin_unlock_irqrestore(&svc->srv_lock, flags);
768 int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
769 int num_threads, char *base_name)
774 for (i = 0; i < num_threads; i++) {
776 sprintf(name, "%s_%02d", base_name, i);
777 rc = ptlrpc_start_thread(dev, svc, name);
779 CERROR("cannot start %s thread #%d: rc %d\n", base_name,
781 ptlrpc_stop_all_threads(svc);
787 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
790 struct l_wait_info lwi = { 0 };
791 struct ptlrpc_svc_data d;
792 struct ptlrpc_thread *thread;
797 OBD_ALLOC(thread, sizeof(*thread));
800 init_waitqueue_head(&thread->t_ctl_waitq);
807 spin_lock_irqsave(&svc->srv_lock, flags);
808 list_add(&thread->t_link, &svc->srv_threads);
809 spin_unlock_irqrestore(&svc->srv_lock, flags);
811 /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
812 * just drop the VM and FILES in ptlrpc_daemonize() right away.
814 rc = kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
816 CERROR("cannot start thread: %d\n", rc);
817 OBD_FREE(thread, sizeof(*thread));
820 l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
826 int ptlrpc_unregister_service(struct ptlrpc_service *service)
831 struct ptlrpc_srv_ni *srv_ni;
832 struct l_wait_info lwi;
833 struct list_head *tmp;
835 LASSERT(list_empty(&service->srv_threads));
837 spin_lock (&ptlrpc_all_services_lock);
838 list_del_init (&service->srv_list);
839 spin_unlock (&ptlrpc_all_services_lock);
841 for (i = 0; i < ptlrpc_ninterfaces; i++) {
842 srv_ni = &service->srv_interfaces[i];
843 CDEBUG(D_NET, "%s: tearing down interface %s\n",
844 service->srv_name, srv_ni->sni_ni->pni_name);
846 /* Unlink all the request buffers. This forces a 'final'
847 * event with its 'unlink' flag set for each posted rqbd */
848 list_for_each(tmp, &srv_ni->sni_active_rqbds) {
849 struct ptlrpc_request_buffer_desc *rqbd =
850 list_entry(tmp, struct ptlrpc_request_buffer_desc,
853 rc = PtlMDUnlink(rqbd->rqbd_md_h);
854 LASSERT (rc == PTL_OK || rc == PTL_INV_MD);
857 /* Wait for the network to release any buffers it's
858 * currently filling */
860 spin_lock_irqsave(&service->srv_lock, flags);
861 rc = srv_ni->sni_nrqbd_receiving;
862 spin_unlock_irqrestore(&service->srv_lock, flags);
867 /* Network access will complete in finite time but
868 * the HUGE timeout lets us CWARN for visibility of
870 lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
871 rc = l_wait_event(service->srv_waitq,
872 srv_ni->sni_nrqbd_receiving == 0,
874 if (rc == -ETIMEDOUT)
875 CWARN("Waiting for request buffers on "
876 "service %s on interface %s ",
877 service->srv_name, srv_ni->sni_ni->pni_name);
880 /* schedule all outstanding replies to terminate them */
881 spin_lock_irqsave(&service->srv_lock, flags);
882 while (!list_empty(&srv_ni->sni_active_replies)) {
883 struct ptlrpc_reply_state *rs =
884 list_entry(srv_ni->sni_active_replies.next,
885 struct ptlrpc_reply_state,
887 ptlrpc_schedule_difficult_reply(rs);
889 spin_unlock_irqrestore(&service->srv_lock, flags);
892 /* purge the request queue. NB No new replies (rqbds all unlinked)
893 * and no service threads, so I'm the only thread noodling the
894 * request queue now */
895 while (!list_empty(&service->srv_request_queue)) {
896 struct ptlrpc_request *req =
897 list_entry(service->srv_request_queue.next,
898 struct ptlrpc_request,
901 list_del(&req->rq_list);
902 service->srv_n_queued_reqs--;
903 service->srv_n_active_reqs++;
905 ptlrpc_server_free_request(service, req);
907 LASSERT(service->srv_n_queued_reqs == 0);
908 LASSERT(service->srv_n_active_reqs == 0);
910 for (i = 0; i < ptlrpc_ninterfaces; i++) {
911 srv_ni = &service->srv_interfaces[i];
912 LASSERT(list_empty(&srv_ni->sni_active_rqbds));
915 /* Now free all the request buffers since nothing references them
917 while (!list_empty(&service->srv_idle_rqbds)) {
918 struct ptlrpc_request_buffer_desc *rqbd =
919 list_entry(service->srv_idle_rqbds.next,
920 struct ptlrpc_request_buffer_desc,
923 ptlrpc_free_rqbd(rqbd);
926 /* wait for all outstanding replies to complete (they were
927 * scheduled having been flagged to abort above) */
928 while (atomic_read(&service->srv_outstanding_replies) != 0) {
929 struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
931 rc = l_wait_event(service->srv_waitq,
932 !list_empty(&service->srv_reply_queue), &lwi);
933 LASSERT(rc == 0 || rc == -ETIMEDOUT);
936 ptlrpc_server_handle_reply(service);
939 CWARN("Unexpectedly long timeout %p\n", service);
942 ptlrpc_lprocfs_unregister_service(service);
945 offsetof(struct ptlrpc_service,
946 srv_interfaces[ptlrpc_ninterfaces]));