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, &srv_ni->sni_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         LASSERT (rs != NULL);
 127         LASSERT (rs->rs_nlocks < RS_MAX_LOCKS);
 128
 129         idx = rs->rs_nlocks++;
 130         rs->rs_locks[idx] = *lock;
 131         rs->rs_modes[idx] = mode;
 132         rs->rs_difficult = 1;
 133 }
 134
 135 void
 136 ptlrpc_schedule_difficult_reply (struct ptlrpc_reply_state *rs)
 137 {
 138         struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
 139
 140 #ifdef CONFIG_SMP
 141         LASSERT (spin_is_locked (&svc->srv_lock));
 142 #endif
 143         LASSERT (rs->rs_difficult);
 144         rs->rs_scheduled_ever = 1;              /* flag any notification attempt */
 145
 146         if (rs->rs_scheduled)                   /* being set up or already notified */
 147                 return;
 148
 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);
 153 }
 154
 155 void
 156 ptlrpc_commit_replies (struct obd_device *obd)
 157 {
 158         struct list_head   *tmp;
 159         struct list_head   *nxt;
 160         unsigned long       flags;
 161
 162         /* Find any replies that have been committed and get their service
 163          * to attend to complete them. */
 164
 165         /* CAVEAT EMPTOR: spinlock ordering!!! */
 166         spin_lock_irqsave (&obd->obd_uncommitted_replies_lock, flags);
 167
 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);
 171
 172                 LASSERT (rs->rs_difficult);
 173
 174                 if (rs->rs_transno <= obd->obd_last_committed) {
 175                         struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
 176
 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);
 181                 }
 182         }
 183
 184         spin_unlock_irqrestore (&obd->obd_uncommitted_replies_lock, flags);
 185 }
 186
 187 static long
 188 timeval_sub(struct timeval *large, struct timeval *small)
 189 {
 190         return (large->tv_sec - small->tv_sec) * 1000000 +
 191                 (large->tv_usec - small->tv_usec);
 192 }
 193
 194 struct ptlrpc_service *
 195 ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size,
 196                 int req_portal, int rep_portal,
 197                 svc_handler_t handler, char *name,
 198                 struct proc_dir_entry *proc_entry)
 199 {
 200         int                                i;
 201         int                                j;
 202         int                                ssize;
 203         struct ptlrpc_service             *service;
 204         struct ptlrpc_srv_ni              *srv_ni;
 205         struct ptlrpc_request_buffer_desc *rqbd;
 206         ENTRY;
 207
 208         LASSERT (ptlrpc_ninterfaces > 0);
 209         LASSERT (nbufs > 0);
 210         LASSERT (bufsize >= max_req_size);
 211
 212         ssize = offsetof (struct ptlrpc_service,
 213                           srv_interfaces[ptlrpc_ninterfaces]);
 214         OBD_ALLOC(service, ssize);
 215         if (service == NULL)
 216                 RETURN(NULL);
 217
 218         service->srv_name = name;
 219         spin_lock_init(&service->srv_lock);
 220         INIT_LIST_HEAD(&service->srv_threads);
 221         init_waitqueue_head(&service->srv_waitq);
 222
 223         service->srv_max_req_size = max_req_size;
 224         service->srv_buf_size = bufsize;
 225         service->srv_rep_portal = rep_portal;
 226         service->srv_req_portal = req_portal;
 227         service->srv_handler = handler;
 228
 229         INIT_LIST_HEAD(&service->srv_request_queue);
 230         INIT_LIST_HEAD(&service->srv_reply_queue);
 231
 232         /* First initialise enough for early teardown */
 233         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 234                 srv_ni = &service->srv_interfaces[i];
 235
 236                 srv_ni->sni_service = service;
 237                 srv_ni->sni_ni = &ptlrpc_interfaces[i];
 238                 INIT_LIST_HEAD(&srv_ni->sni_rqbds);
 239                 INIT_LIST_HEAD(&srv_ni->sni_active_replies);
 240         }
 241
 242         spin_lock (&ptlrpc_all_services_lock);
 243         list_add (&service->srv_list, &ptlrpc_all_services);
 244         spin_unlock (&ptlrpc_all_services_lock);
 245
 246         /* Now allocate the request buffers, assuming all interfaces require
 247          * the same number. */
 248         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 249                 srv_ni = &service->srv_interfaces[i];
 250                 CDEBUG (D_NET, "%s: initialising interface %s\n", name,
 251                         srv_ni->sni_ni->pni_name);
 252
 253                 for (j = 0; j < nbufs; j++) {
 254                         rqbd = ptlrpc_alloc_rqbd (srv_ni);
 255
 256                         if (rqbd == NULL) {
 257                                 CERROR ("%s.%d: Can't allocate request %d "
 258                                         "on %s\n", name, i, j,
 259                                         srv_ni->sni_ni->pni_name);
 260                                 GOTO(failed, NULL);
 261                         }
 262                         ptlrpc_register_rqbd (rqbd);
 263                 }
 264         }
 265
 266         if (proc_entry != NULL)
 267                 ptlrpc_lprocfs_register_service(proc_entry, service);
 268
 269         CDEBUG(D_NET, "%s: Started on %d interfaces, listening on portal %d\n",
 270                service->srv_name, ptlrpc_ninterfaces, service->srv_req_portal);
 271
 272         RETURN(service);
 273 failed:
 274         ptlrpc_unregister_service(service);
 275         return NULL;
 276 }
 277
 278 static int
 279 ptlrpc_server_handle_request (struct ptlrpc_service *svc)
 280 {
 281         struct ptlrpc_request *request;
 282         unsigned long          flags;
 283         struct timeval         work_start;
 284         struct timeval         work_end;
 285         long                   timediff;
 286         int                    refcount;
 287         int                    rc;
 288         ENTRY;
 289
 290         spin_lock_irqsave (&svc->srv_lock, flags);
 291         if (list_empty (&svc->srv_request_queue) ||
 292             (svc->srv_n_difficult_replies != 0 &&
 293              svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
 294                 /* If all the other threads are handling requests, I must
 295                  * remain free to handle any 'difficult' reply that might
 296                  * block them */
 297                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 298                 RETURN(0);
 299         }
 300
 301         request = list_entry (svc->srv_request_queue.next,
 302                               struct ptlrpc_request, rq_list);
 303         list_del_init (&request->rq_list);
 304         svc->srv_n_queued_reqs--;
 305         svc->srv_n_active_reqs++;
 306
 307         spin_unlock_irqrestore (&svc->srv_lock, flags);
 308
 309         do_gettimeofday(&work_start);
 310         timediff = timeval_sub(&work_start, &request->rq_arrival_time);
 311         if (svc->srv_stats != NULL) {
 312                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
 313                                     timediff);
 314                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
 315                                     svc->srv_n_queued_reqs);
 316                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
 317                                     svc->srv_n_active_reqs);
 318         }
 319
 320 #if SWAB_PARANOIA
 321         /* Clear request swab mask; this is a new request */
 322         request->rq_req_swab_mask = 0;
 323 #endif
 324         rc = lustre_unpack_msg (request->rq_reqmsg, request->rq_reqlen);
 325         if (rc != 0) {
 326                 CERROR ("error unpacking request: ptl %d from "LPX64
 327                         " xid "LPU64"\n", svc->srv_req_portal,
 328                        request->rq_peer.peer_nid, request->rq_xid);
 329                 goto out;
 330         }
 331
 332         rc = -EINVAL;
 333         if (request->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
 334                 CERROR("wrong packet type received (type=%u) from "
 335                        LPX64"\n", request->rq_reqmsg->type,
 336                        request->rq_peer.peer_nid);
 337                 goto out;
 338         }
 339
 340         CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
 341
 342         /* Discard requests queued for longer than my timeout.  If the
 343          * client's timeout is similar to mine, she'll be timing out this
 344          * REQ anyway (bug 1502) */
 345         if (timediff / 1000000 > (long)obd_timeout) {
 346                 CERROR("Dropping timed-out request from "LPX64
 347                        ": %ld seconds old\n",
 348                        request->rq_peer.peer_nid, timediff / 1000000);
 349                 goto out;
 350         }
 351
 352         request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
 353
 354         if (request->rq_export) {
 355                 if (request->rq_reqmsg->conn_cnt <
 356                     request->rq_export->exp_conn_cnt) {
 357                         DEBUG_REQ(D_ERROR, request,
 358                                   "DROPPING req from old connection %d < %d",
 359                                   request->rq_reqmsg->conn_cnt,
 360                                   request->rq_export->exp_conn_cnt);
 361                         goto put_conn;
 362                 }
 363
 364                 request->rq_export->exp_last_request_time =
 365                         LTIME_S(CURRENT_TIME);
 366         }
 367
 368         CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 369                "%s:%s+%d:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 370                (request->rq_export ?
 371                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 372                (request->rq_export ?
 373                 atomic_read(&request->rq_export->exp_refcount) : -99),
 374                request->rq_reqmsg->status, request->rq_xid,
 375                request->rq_peer.peer_ni->pni_name,
 376                request->rq_peer.peer_nid,
 377                request->rq_reqmsg->opc);
 378
 379         rc = svc->srv_handler(request);
 380         CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 381                "%s:%s+%d:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 382                (request->rq_export ?
 383                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 384                (request->rq_export ?
 385                 atomic_read(&request->rq_export->exp_refcount) : -99),
 386                request->rq_reqmsg->status, request->rq_xid,
 387                request->rq_peer.peer_ni->pni_name,
 388                request->rq_peer.peer_nid,
 389                request->rq_reqmsg->opc);
 390
 391 put_conn:
 392         if (request->rq_export != NULL)
 393                 class_export_put(request->rq_export);
 394
 395  out:
 396         do_gettimeofday(&work_end);
 397
 398         timediff = timeval_sub(&work_end, &work_start);
 399
 400         CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
 401                "request "LPU64" opc %u from NID "LPX64" processed in %ldus "
 402                "(%ldus total)\n", request->rq_xid, request->rq_reqmsg->opc,
 403                request->rq_peer.peer_nid,
 404                timediff, timeval_sub(&work_end, &request->rq_arrival_time));
 405
 406         if (svc->srv_stats != NULL) {
 407                 int opc = opcode_offset(request->rq_reqmsg->opc);
 408                 if (opc > 0) {
 409                         LASSERT(opc < LUSTRE_MAX_OPCODES);
 410                         lprocfs_counter_add(svc->srv_stats,
 411                                             opc + PTLRPC_LAST_CNTR,
 412                                             timediff);
 413                 }
 414         }
 415
 416         spin_lock_irqsave(&svc->srv_lock, flags);
 417         svc->srv_n_active_reqs--;
 418         refcount = --(request->rq_rqbd->rqbd_refcount);
 419         spin_unlock_irqrestore(&svc->srv_lock, flags);
 420
 421         if (refcount == 0) {
 422                 /* rqbd now idle: repost */
 423                 ptlrpc_register_rqbd(request->rq_rqbd);
 424         }
 425
 426         ptlrpc_free_server_req(request);
 427
 428         RETURN(1);
 429 }
 430
 431 static int
 432 ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
 433 {
 434         struct ptlrpc_reply_state *rs;
 435         unsigned long              flags;
 436         struct obd_export         *exp;
 437         struct obd_device         *obd;
 438         int                        nlocks;
 439         int                        been_handled;
 440         ENTRY;
 441
 442         spin_lock_irqsave (&svc->srv_lock, flags);
 443         if (list_empty (&svc->srv_reply_queue)) {
 444                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 445                 RETURN(0);
 446         }
 447
 448         rs = list_entry (svc->srv_reply_queue.next,
 449                          struct ptlrpc_reply_state, rs_list);
 450
 451         exp = rs->rs_export;
 452         obd = exp->exp_obd;
 453
 454         LASSERT (rs->rs_difficult);
 455         LASSERT (rs->rs_scheduled);
 456
 457         list_del_init (&rs->rs_list);
 458
 459         /* Disengage from notifiers carefully (lock ordering!) */
 460         spin_unlock(&svc->srv_lock);
 461
 462         spin_lock (&obd->obd_uncommitted_replies_lock);
 463         /* Noop if removed already */
 464         list_del_init (&rs->rs_obd_list);
 465         spin_unlock (&obd->obd_uncommitted_replies_lock);
 466
 467         spin_lock (&exp->exp_lock);
 468         /* Noop if removed already */
 469         list_del_init (&rs->rs_exp_list);
 470         spin_unlock (&exp->exp_lock);
 471
 472         spin_lock(&svc->srv_lock);
 473
 474         been_handled = rs->rs_handled;
 475         rs->rs_handled = 1;
 476
 477         nlocks = rs->rs_nlocks;                 /* atomic "steal", but */
 478         rs->rs_nlocks = 0;                      /* locks still on rs_locks! */
 479
 480         if (nlocks == 0 && !been_handled) {
 481                 /* If we see this, we should already have seen the warning
 482                  * in mds_steal_ack_locks()  */
 483                 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
 484                       " o%d NID"LPX64"\n",
 485                       rs,
 486                       rs->rs_xid, rs->rs_transno,
 487                       rs->rs_msg.opc, exp->exp_connection->c_peer.peer_nid);
 488         }
 489
 490         if ((!been_handled && rs->rs_on_net) ||
 491             nlocks > 0) {
 492                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 493
 494                 if (!been_handled && rs->rs_on_net) {
 495                         PtlMDUnlink(rs->rs_md_h);
 496                         /* Ignore return code; we're racing with
 497                          * completion... */
 498                 }
 499
 500                 while (nlocks-- > 0)
 501                         ldlm_lock_decref(&rs->rs_locks[nlocks],
 502                                          rs->rs_modes[nlocks]);
 503
 504                 spin_lock_irqsave(&svc->srv_lock, flags);
 505         }
 506
 507         rs->rs_scheduled = 0;
 508
 509         if (!rs->rs_on_net) {
 510                 /* Off the net */
 511                 svc->srv_n_difficult_replies--;
 512                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 513
 514                 class_export_put (exp);
 515                 rs->rs_export = NULL;
 516                 lustre_free_reply_state (rs);
 517                 atomic_dec (&svc->srv_outstanding_replies);
 518                 RETURN(1);
 519         }
 520
 521         /* still on the net; callback will schedule */
 522         spin_unlock_irqrestore (&svc->srv_lock, flags);
 523         RETURN(1);
 524 }
 525
 526 #ifndef __KERNEL__
 527 /* FIXME make use of timeout later */
 528 int
 529 liblustre_check_services (void *arg)
 530 {
 531         int  did_something = 0;
 532         struct list_head *tmp, *nxt;
 533         ENTRY;
 534
 535         /* I'm relying on being single threaded, not to have to lock
 536          * ptlrpc_all_services etc */
 537         list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
 538                 struct ptlrpc_service *svc =
 539                         list_entry (tmp, struct ptlrpc_service, srv_list);
 540
 541                 if (svc->srv_nthreads != 0)     /* I've recursed */
 542                         continue;
 543
 544                 /* service threads can block for bulk, so this limits us
 545                  * (arbitrarily) to recursing 1 stack frame per service.
 546                  * Note that the problem with recursion is that we have to
 547                  * unwind completely before our caller can resume. */
 548
 549                 svc->srv_nthreads++;
 550
 551                 while (ptlrpc_server_handle_reply (svc))
 552                         did_something++;
 553
 554                 while (ptlrpc_server_handle_request (svc))
 555                         did_something++;
 556
 557                 svc->srv_nthreads--;
 558         }
 559
 560         RETURN(did_something);
 561 }
 562
 563 #else /* __KERNEL__ */
 564
 565 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
 566 void ptlrpc_daemonize(void)
 567 {
 568         exit_mm(current);
 569         lustre_daemonize_helper();
 570         exit_files(current);
 571         reparent_to_init();
 572 }
 573
 574 static int ptlrpc_main(void *arg)
 575 {
 576         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
 577         struct ptlrpc_service  *svc = data->svc;
 578         struct ptlrpc_thread   *thread = data->thread;
 579         unsigned long           flags;
 580         ENTRY;
 581
 582         lock_kernel();
 583         ptlrpc_daemonize();
 584
 585         SIGNAL_MASK_LOCK(current, flags);
 586         sigfillset(&current->blocked);
 587         RECALC_SIGPENDING;
 588         SIGNAL_MASK_UNLOCK(current, flags);
 589
 590         THREAD_NAME(current->comm, "%s", data->name);
 591         unlock_kernel();
 592
 593         /* Record that the thread is running */
 594         thread->t_flags = SVC_RUNNING;
 595         wake_up(&thread->t_ctl_waitq);
 596
 597         spin_lock_irqsave(&svc->srv_lock, flags);
 598         svc->srv_nthreads++;
 599         spin_unlock_irqrestore(&svc->srv_lock, flags);
 600
 601         /* XXX maintain a list of all managed devices: insert here */
 602
 603         while ((thread->t_flags & SVC_STOPPING) == 0 ||
 604                svc->srv_n_difficult_replies != 0) {
 605                 /* Don't exit while there are replies to be handled */
 606                 struct l_wait_info lwi = { 0 };
 607
 608                 l_wait_event_exclusive (svc->srv_waitq,
 609                               (thread->t_flags & SVC_STOPPING) != 0 ||
 610                               !list_empty (&svc->srv_reply_queue) ||
 611                               (!list_empty (&svc->srv_request_queue) &&
 612                                (svc->srv_n_difficult_replies == 0 ||
 613                                 svc->srv_n_active_reqs <
 614                                 (svc->srv_nthreads - 1))),
 615                               &lwi);
 616
 617                 if (!list_empty (&svc->srv_reply_queue))
 618                         ptlrpc_server_handle_reply (svc);
 619
 620                 /* only handle requests if there are no difficult replies
 621                  * outstanding, or I'm not the last thread handling
 622                  * requests */
 623                 if (!list_empty (&svc->srv_request_queue) &&
 624                     (svc->srv_n_difficult_replies == 0 ||
 625                      svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
 626                         ptlrpc_server_handle_request (svc);
 627         }
 628
 629         spin_lock_irqsave(&svc->srv_lock, flags);
 630
 631         svc->srv_nthreads--;                    /* must know immediately */
 632         thread->t_flags = SVC_STOPPED;
 633         wake_up(&thread->t_ctl_waitq);
 634
 635         spin_unlock_irqrestore(&svc->srv_lock, flags);
 636
 637         CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
 638         return 0;
 639 }
 640
 641 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
 642                                struct ptlrpc_thread *thread)
 643 {
 644         struct l_wait_info lwi = { 0 };
 645         unsigned long      flags;
 646
 647         spin_lock_irqsave(&svc->srv_lock, flags);
 648         thread->t_flags = SVC_STOPPING;
 649         spin_unlock_irqrestore(&svc->srv_lock, flags);
 650
 651         wake_up_all(&svc->srv_waitq);
 652         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
 653                      &lwi);
 654
 655         spin_lock_irqsave(&svc->srv_lock, flags);
 656         list_del(&thread->t_link);
 657         spin_unlock_irqrestore(&svc->srv_lock, flags);
 658
 659         OBD_FREE(thread, sizeof(*thread));
 660 }
 661
 662 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
 663 {
 664         unsigned long flags;
 665         struct ptlrpc_thread *thread;
 666
 667         spin_lock_irqsave(&svc->srv_lock, flags);
 668         while (!list_empty(&svc->srv_threads)) {
 669                 thread = list_entry(svc->srv_threads.next,
 670                                     struct ptlrpc_thread, t_link);
 671
 672                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 673                 ptlrpc_stop_thread(svc, thread);
 674                 spin_lock_irqsave(&svc->srv_lock, flags);
 675         }
 676
 677         spin_unlock_irqrestore(&svc->srv_lock, flags);
 678 }
 679
 680 int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
 681                            int num_threads, char *base_name)
 682 {
 683         int i, rc = 0;
 684         ENTRY;
 685
 686         for (i = 0; i < num_threads; i++) {
 687                 char name[32];
 688                 sprintf(name, "%s_%02d", base_name, i);
 689                 rc = ptlrpc_start_thread(dev, svc, name);
 690                 if (rc) {
 691                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
 692                                i, rc);
 693                         ptlrpc_stop_all_threads(svc);
 694                 }
 695         }
 696         RETURN(rc);
 697 }
 698
 699 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
 700                         char *name)
 701 {
 702         struct l_wait_info lwi = { 0 };
 703         struct ptlrpc_svc_data d;
 704         struct ptlrpc_thread *thread;
 705         unsigned long flags;
 706         int rc;
 707         ENTRY;
 708
 709         OBD_ALLOC(thread, sizeof(*thread));
 710         if (thread == NULL)
 711                 RETURN(-ENOMEM);
 712         init_waitqueue_head(&thread->t_ctl_waitq);
 713
 714         d.dev = dev;
 715         d.svc = svc;
 716         d.name = name;
 717         d.thread = thread;
 718
 719         spin_lock_irqsave(&svc->srv_lock, flags);
 720         list_add(&thread->t_link, &svc->srv_threads);
 721         spin_unlock_irqrestore(&svc->srv_lock, flags);
 722
 723         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
 724          * just drop the VM and FILES in ptlrpc_daemonize() right away.
 725          */
 726         rc = kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
 727         if (rc < 0) {
 728                 CERROR("cannot start thread: %d\n", rc);
 729                 OBD_FREE(thread, sizeof(*thread));
 730                 RETURN(rc);
 731         }
 732         l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
 733
 734         RETURN(0);
 735 }
 736 #endif
 737
 738 int ptlrpc_unregister_service(struct ptlrpc_service *service)
 739 {
 740         int                   i;
 741         int                   rc;
 742         unsigned long         flags;
 743         struct ptlrpc_srv_ni *srv_ni;
 744         struct l_wait_info    lwi;
 745         struct list_head     *tmp;
 746
 747         LASSERT(list_empty(&service->srv_threads));
 748
 749         spin_lock (&ptlrpc_all_services_lock);
 750         list_del_init (&service->srv_list);
 751         spin_unlock (&ptlrpc_all_services_lock);
 752
 753         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 754                 srv_ni = &service->srv_interfaces[i];
 755                 CDEBUG(D_NET, "%s: tearing down interface %s\n",
 756                        service->srv_name, srv_ni->sni_ni->pni_name);
 757
 758                 /* Unlink all the request buffers.  This forces a 'final'
 759                  * event with its 'unlink' flag set for each rqbd */
 760                 list_for_each(tmp, &srv_ni->sni_rqbds) {
 761                         struct ptlrpc_request_buffer_desc *rqbd =
 762                                 list_entry(tmp, struct ptlrpc_request_buffer_desc,
 763                                            rqbd_list);
 764
 765                         rc = PtlMDUnlink(rqbd->rqbd_md_h);
 766                         LASSERT (rc == PTL_OK || rc == PTL_INV_MD);
 767                 }
 768
 769                 /* Wait for the network to release any buffers it's
 770                  * currently filling */
 771                 for (;;) {
 772                         spin_lock_irqsave(&service->srv_lock, flags);
 773                         rc = srv_ni->sni_nrqbd_receiving;
 774                         spin_unlock_irqrestore(&service->srv_lock, flags);
 775
 776                         if (rc == 0)
 777                                 break;
 778
 779                         /* Network access will complete in finite time but
 780                          * the HUGE timeout lets us CWARN for visibility of
 781                          * sluggish NALs */
 782                         lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
 783                         rc = l_wait_event(service->srv_waitq,
 784                                           srv_ni->sni_nrqbd_receiving == 0,
 785                                           &lwi);
 786                         if (rc == -ETIMEDOUT)
 787                                 CWARN("Waiting for request buffers on "
 788                                       "service %s on interface %s ",
 789                                       service->srv_name, srv_ni->sni_ni->pni_name);
 790                 }
 791
 792                 /* schedule all outstanding replies to terminate them */
 793                 spin_lock_irqsave(&service->srv_lock, flags);
 794                 while (!list_empty(&srv_ni->sni_active_replies)) {
 795                         struct ptlrpc_reply_state *rs =
 796                                 list_entry(srv_ni->sni_active_replies.next,
 797                                            struct ptlrpc_reply_state,
 798                                            rs_list);
 799                         ptlrpc_schedule_difficult_reply(rs);
 800                 }
 801                 spin_unlock_irqrestore(&service->srv_lock, flags);
 802         }
 803
 804         /* purge the request queue.  NB No new replies (rqbds all unlinked)
 805          * and no service threads, so I'm the only thread noodling the
 806          * request queue now */
 807         while (!list_empty(&service->srv_request_queue)) {
 808                 struct ptlrpc_request *req =
 809                         list_entry(service->srv_request_queue.next,
 810                                    struct ptlrpc_request,
 811                                    rq_list);
 812
 813                 list_del(&req->rq_list);
 814                 service->srv_n_queued_reqs--;
 815                 req->rq_rqbd->rqbd_refcount--;
 816
 817                 ptlrpc_free_server_req(req);
 818         }
 819         LASSERT(service->srv_n_queued_reqs == 0);
 820
 821         /* Now free all the request buffers since nothing references them
 822          * any more... */
 823         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 824                 srv_ni = &service->srv_interfaces[i];
 825
 826                 while (!list_empty(&srv_ni->sni_rqbds)) {
 827                         struct ptlrpc_request_buffer_desc *rqbd =
 828                                 list_entry(srv_ni->sni_rqbds.next,
 829                                            struct ptlrpc_request_buffer_desc,
 830                                            rqbd_list);
 831
 832                         ptlrpc_free_rqbd(rqbd);
 833                 }
 834         }
 835
 836         /* wait for all outstanding replies to complete (they were
 837          * scheduled having been flagged to abort above) */
 838         while (atomic_read(&service->srv_outstanding_replies) != 0) {
 839                 struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
 840
 841                 rc = l_wait_event(service->srv_waitq,
 842                                   !list_empty(&service->srv_reply_queue), &lwi);
 843                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
 844
 845                 if (rc == 0) {
 846                         ptlrpc_server_handle_reply(service);
 847                         continue;
 848                 }
 849                 CWARN("Unexpectedly long timeout %p\n", service);
 850         }
 851
 852         ptlrpc_lprocfs_unregister_service(service);
 853
 854         OBD_FREE(service,
 855                  offsetof(struct ptlrpc_service,
 856                           srv_interfaces[ptlrpc_ninterfaces]));
 857         return 0;
 858 }