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 obd_export     *export = NULL;
 361         struct ptlrpc_request *request;
 362         unsigned long          flags;
 363         struct timeval         work_start;
 364         struct timeval         work_end;
 365         long                   timediff;
 366         int                    rc;
 367         char                   str[PTL_NALFMT_SIZE];
 368         ENTRY;
 369
 370         spin_lock_irqsave (&svc->srv_lock, flags);
 371         if (list_empty (&svc->srv_request_queue) ||
 372             (svc->srv_n_difficult_replies != 0 &&
 373              svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
 374                 /* If all the other threads are handling requests, I must
 375                  * remain free to handle any 'difficult' reply that might
 376                  * block them */
 377                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 378                 RETURN(0);
 379         }
 380
 381         request = list_entry (svc->srv_request_queue.next,
 382                               struct ptlrpc_request, rq_list);
 383         list_del_init (&request->rq_list);
 384         svc->srv_n_queued_reqs--;
 385         svc->srv_n_active_reqs++;
 386
 387         spin_unlock_irqrestore (&svc->srv_lock, flags);
 388
 389         do_gettimeofday(&work_start);
 390         timediff = timeval_sub(&work_start, &request->rq_arrival_time);
 391         if (svc->srv_stats != NULL) {
 392                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
 393                                     timediff);
 394                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
 395                                     svc->srv_n_queued_reqs);
 396                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
 397                                     svc->srv_n_active_reqs);
 398         }
 399
 400 #if SWAB_PARANOIA
 401         /* Clear request swab mask; this is a new request */
 402         request->rq_req_swab_mask = 0;
 403 #endif
 404         rc = lustre_unpack_msg (request->rq_reqmsg, request->rq_reqlen);
 405         if (rc != 0) {
 406                 CERROR ("error unpacking request: ptl %d from %s"
 407                         " xid "LPU64"\n", svc->srv_req_portal,
 408                         ptlrpc_peernid2str(&request->rq_peer, str),
 409                        request->rq_xid);
 410                 goto out;
 411         }
 412
 413         rc = -EINVAL;
 414         if (request->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
 415                 CERROR("wrong packet type received (type=%u) from %s\n",
 416                        request->rq_reqmsg->type,
 417                        ptlrpc_peernid2str(&request->rq_peer, str));
 418                 goto out;
 419         }
 420
 421         CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
 422
 423         /* Discard requests queued for longer than my timeout.  If the
 424          * client's timeout is similar to mine, she'll be timing out this
 425          * REQ anyway (bug 1502) */
 426         if (timediff / 1000000 > (long)obd_timeout) {
 427                 CERROR("Dropping timed-out request from %s: %ld seconds old\n",
 428                        ptlrpc_peernid2str(&request->rq_peer, str),
 429                        timediff / 1000000);
 430                 goto out;
 431         }
 432
 433         request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
 434
 435         if (request->rq_export) {
 436                 if (request->rq_reqmsg->conn_cnt <
 437                     request->rq_export->exp_conn_cnt) {
 438                         DEBUG_REQ(D_ERROR, request,
 439                                   "DROPPING req from old connection %d < %d",
 440                                   request->rq_reqmsg->conn_cnt,
 441                                   request->rq_export->exp_conn_cnt);
 442                         goto put_conn;
 443                 }
 444
 445                 export = class_export_rpc_get(request->rq_export);
 446                 request->rq_export->exp_last_request_time =
 447                         LTIME_S(CURRENT_TIME);
 448         }
 449
 450         CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 451                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
 452                (request->rq_export ?
 453                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 454                (request->rq_export ?
 455                 atomic_read(&request->rq_export->exp_refcount) : -99),
 456                request->rq_reqmsg->status, request->rq_xid,
 457                request->rq_peer.peer_ni->pni_name,
 458                ptlrpc_peernid2str(&request->rq_peer, str),
 459                request->rq_reqmsg->opc);
 460
 461         rc = svc->srv_handler(request);
 462         CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 463                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
 464                (request->rq_export ?
 465                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 466                (request->rq_export ?
 467                 atomic_read(&request->rq_export->exp_refcount) : -99),
 468                request->rq_reqmsg->status, request->rq_xid,
 469                request->rq_peer.peer_ni->pni_name,
 470                ptlrpc_peernid2str(&request->rq_peer, str),
 471                request->rq_reqmsg->opc);
 472
 473         if (export != NULL)
 474                 class_export_rpc_put(export);
 475
 476 put_conn:
 477         if (request->rq_export != NULL)
 478                 class_export_put(request->rq_export);
 479
 480  out:
 481         do_gettimeofday(&work_end);
 482
 483         timediff = timeval_sub(&work_end, &work_start);
 484
 485         CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
 486                "request "LPU64" opc %u from NID %s processed in %ldus "
 487                "(%ldus total)\n", request->rq_xid, request->rq_reqmsg->opc,
 488                ptlrpc_peernid2str(&request->rq_peer, str),
 489                timediff, timeval_sub(&work_end, &request->rq_arrival_time));
 490
 491         if (svc->srv_stats != NULL) {
 492                 int opc = opcode_offset(request->rq_reqmsg->opc);
 493                 if (opc > 0) {
 494                         LASSERT(opc < LUSTRE_MAX_OPCODES);
 495                         lprocfs_counter_add(svc->srv_stats,
 496                                             opc + PTLRPC_LAST_CNTR,
 497                                             timediff);
 498                 }
 499         }
 500
 501         ptlrpc_server_free_request(svc, request);
 502
 503         RETURN(1);
 504 }
 505
 506 static int
 507 ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
 508 {
 509         struct ptlrpc_reply_state *rs;
 510         unsigned long              flags;
 511         struct obd_export         *exp;
 512         struct obd_device         *obd;
 513         int                        nlocks;
 514         int                        been_handled;
 515         char                       str[PTL_NALFMT_SIZE];
 516         ENTRY;
 517
 518         spin_lock_irqsave (&svc->srv_lock, flags);
 519         if (list_empty (&svc->srv_reply_queue)) {
 520                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 521                 RETURN(0);
 522         }
 523
 524         rs = list_entry (svc->srv_reply_queue.next,
 525                          struct ptlrpc_reply_state, rs_list);
 526
 527         exp = rs->rs_export;
 528         obd = exp->exp_obd;
 529
 530         LASSERT (rs->rs_difficult);
 531         LASSERT (rs->rs_scheduled);
 532
 533         list_del_init (&rs->rs_list);
 534
 535         /* Disengage from notifiers carefully (lock ordering!) */
 536         spin_unlock(&svc->srv_lock);
 537
 538         spin_lock (&obd->obd_uncommitted_replies_lock);
 539         /* Noop if removed already */
 540         list_del_init (&rs->rs_obd_list);
 541         spin_unlock (&obd->obd_uncommitted_replies_lock);
 542
 543         spin_lock (&exp->exp_lock);
 544         /* Noop if removed already */
 545         list_del_init (&rs->rs_exp_list);
 546         spin_unlock (&exp->exp_lock);
 547
 548         spin_lock(&svc->srv_lock);
 549
 550         been_handled = rs->rs_handled;
 551         rs->rs_handled = 1;
 552
 553         nlocks = rs->rs_nlocks;                 /* atomic "steal", but */
 554         rs->rs_nlocks = 0;                      /* locks still on rs_locks! */
 555
 556         if (nlocks == 0 && !been_handled) {
 557                 /* If we see this, we should already have seen the warning
 558                  * in mds_steal_ack_locks()  */
 559                 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
 560                       " o%d NID %s\n",
 561                       rs,
 562                       rs->rs_xid, rs->rs_transno,
 563                       rs->rs_msg.opc,
 564                       ptlrpc_peernid2str(&exp->exp_connection->c_peer, str));
 565         }
 566
 567         if ((!been_handled && rs->rs_on_net) ||
 568             nlocks > 0) {
 569                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 570
 571                 if (!been_handled && rs->rs_on_net) {
 572                         PtlMDUnlink(rs->rs_md_h);
 573                         /* Ignore return code; we're racing with
 574                          * completion... */
 575                 }
 576
 577                 while (nlocks-- > 0)
 578                         ldlm_lock_decref(&rs->rs_locks[nlocks],
 579                                          rs->rs_modes[nlocks]);
 580
 581                 spin_lock_irqsave(&svc->srv_lock, flags);
 582         }
 583
 584         rs->rs_scheduled = 0;
 585
 586         if (!rs->rs_on_net) {
 587                 /* Off the net */
 588                 svc->srv_n_difficult_replies--;
 589                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 590
 591                 class_export_put (exp);
 592                 rs->rs_export = NULL;
 593                 lustre_free_reply_state (rs);
 594                 atomic_dec (&svc->srv_outstanding_replies);
 595                 RETURN(1);
 596         }
 597
 598         /* still on the net; callback will schedule */
 599         spin_unlock_irqrestore (&svc->srv_lock, flags);
 600         RETURN(1);
 601 }
 602
 603 #ifndef __KERNEL__
 604 /* FIXME make use of timeout later */
 605 int
 606 liblustre_check_services (void *arg)
 607 {
 608         int  did_something = 0;
 609         int  rc;
 610         struct list_head *tmp, *nxt;
 611         ENTRY;
 612
 613         /* I'm relying on being single threaded, not to have to lock
 614          * ptlrpc_all_services etc */
 615         list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
 616                 struct ptlrpc_service *svc =
 617                         list_entry (tmp, struct ptlrpc_service, srv_list);
 618
 619                 if (svc->srv_nthreads != 0)     /* I've recursed */
 620                         continue;
 621
 622                 /* service threads can block for bulk, so this limits us
 623                  * (arbitrarily) to recursing 1 stack frame per service.
 624                  * Note that the problem with recursion is that we have to
 625                  * unwind completely before our caller can resume. */
 626
 627                 svc->srv_nthreads++;
 628
 629                 do {
 630                         rc = ptlrpc_server_handle_reply(svc);
 631                         rc |= ptlrpc_server_handle_request(svc);
 632                         rc |= (ptlrpc_server_post_idle_rqbds(svc) > 0);
 633                         did_something |= rc;
 634                 } while (rc);
 635
 636                 svc->srv_nthreads--;
 637         }
 638
 639         RETURN(did_something);
 640 }
 641
 642 #else /* __KERNEL__ */
 643
 644 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
 645 void ptlrpc_daemonize(void)
 646 {
 647         exit_mm(current);
 648         lustre_daemonize_helper();
 649         exit_files(current);
 650         reparent_to_init();
 651 }
 652
 653 static int
 654 ptlrpc_retry_rqbds(void *arg)
 655 {
 656         struct ptlrpc_service *svc = (struct ptlrpc_service *)arg;
 657
 658         svc->srv_rqbd_timeout = 0;
 659         return (-ETIMEDOUT);
 660 }
 661
 662 static int ptlrpc_main(void *arg)
 663 {
 664         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
 665         struct ptlrpc_service  *svc = data->svc;
 666         struct ptlrpc_thread   *thread = data->thread;
 667         unsigned long           flags;
 668         ENTRY;
 669
 670         lock_kernel();
 671         ptlrpc_daemonize();
 672
 673         SIGNAL_MASK_LOCK(current, flags);
 674         sigfillset(&current->blocked);
 675         RECALC_SIGPENDING;
 676         SIGNAL_MASK_UNLOCK(current, flags);
 677
 678         LASSERTF(strlen(data->name) < sizeof(current->comm),
 679                  "name %d > len %d\n",
 680                  (int)strlen(data->name), (int)sizeof(current->comm));
 681         THREAD_NAME(current->comm, sizeof(current->comm) - 1, "%s", data->name);
 682
 683         unlock_kernel();
 684
 685         /* Record that the thread is running */
 686         thread->t_flags = SVC_RUNNING;
 687         wake_up(&thread->t_ctl_waitq);
 688
 689         spin_lock_irqsave(&svc->srv_lock, flags);
 690         svc->srv_nthreads++;
 691         spin_unlock_irqrestore(&svc->srv_lock, flags);
 692
 693         /* XXX maintain a list of all managed devices: insert here */
 694
 695         while ((thread->t_flags & SVC_STOPPING) == 0 ||
 696                svc->srv_n_difficult_replies != 0) {
 697                 /* Don't exit while there are replies to be handled */
 698                 struct l_wait_info lwi = LWI_TIMEOUT(svc->srv_rqbd_timeout,
 699                                                      ptlrpc_retry_rqbds, svc);
 700
 701                 l_wait_event_exclusive (svc->srv_waitq,
 702                               ((thread->t_flags & SVC_STOPPING) != 0 &&
 703                                svc->srv_n_difficult_replies == 0) ||
 704                               (!list_empty(&svc->srv_idle_rqbds) &&
 705                                svc->srv_rqbd_timeout == 0) ||
 706                               !list_empty (&svc->srv_reply_queue) ||
 707                               (!list_empty (&svc->srv_request_queue) &&
 708                                (svc->srv_n_difficult_replies == 0 ||
 709                                 svc->srv_n_active_reqs <
 710                                 (svc->srv_nthreads - 1))),
 711                               &lwi);
 712
 713                 if (!list_empty (&svc->srv_reply_queue))
 714                         ptlrpc_server_handle_reply (svc);
 715
 716                 /* only handle requests if there are no difficult replies
 717                  * outstanding, or I'm not the last thread handling
 718                  * requests */
 719                 if (!list_empty (&svc->srv_request_queue) &&
 720                     (svc->srv_n_difficult_replies == 0 ||
 721                      svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
 722                         ptlrpc_server_handle_request (svc);
 723
 724                 if (!list_empty(&svc->srv_idle_rqbds) &&
 725                     ptlrpc_server_post_idle_rqbds(svc) < 0) {
 726                         /* I just failed to repost request buffers.  Wait
 727                          * for a timeout (unless something else happens)
 728                          * before I try again */
 729                         svc->srv_rqbd_timeout = HZ/10;
 730                 }
 731         }
 732
 733         spin_lock_irqsave(&svc->srv_lock, flags);
 734
 735         svc->srv_nthreads--;                    /* must know immediately */
 736         thread->t_flags = SVC_STOPPED;
 737         wake_up(&thread->t_ctl_waitq);
 738
 739         spin_unlock_irqrestore(&svc->srv_lock, flags);
 740
 741         CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
 742         return 0;
 743 }
 744
 745 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
 746                                struct ptlrpc_thread *thread)
 747 {
 748         struct l_wait_info lwi = { 0 };
 749         unsigned long      flags;
 750
 751         spin_lock_irqsave(&svc->srv_lock, flags);
 752         thread->t_flags = SVC_STOPPING;
 753         spin_unlock_irqrestore(&svc->srv_lock, flags);
 754
 755         wake_up_all(&svc->srv_waitq);
 756         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
 757                      &lwi);
 758
 759         spin_lock_irqsave(&svc->srv_lock, flags);
 760         list_del(&thread->t_link);
 761         spin_unlock_irqrestore(&svc->srv_lock, flags);
 762
 763         OBD_FREE(thread, sizeof(*thread));
 764 }
 765
 766 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
 767 {
 768         unsigned long flags;
 769         struct ptlrpc_thread *thread;
 770
 771         spin_lock_irqsave(&svc->srv_lock, flags);
 772         while (!list_empty(&svc->srv_threads)) {
 773                 thread = list_entry(svc->srv_threads.next,
 774                                     struct ptlrpc_thread, t_link);
 775
 776                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 777                 ptlrpc_stop_thread(svc, thread);
 778                 spin_lock_irqsave(&svc->srv_lock, flags);
 779         }
 780
 781         spin_unlock_irqrestore(&svc->srv_lock, flags);
 782 }
 783
 784 /* @base_name should be 12 characters or less - 3 will be added on */
 785 int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
 786                            int num_threads, char *base_name)
 787 {
 788         int i, rc = 0;
 789         ENTRY;
 790
 791         for (i = 0; i < num_threads; i++) {
 792                 char name[32];
 793                 sprintf(name, "%s_%02d", base_name, i);
 794                 rc = ptlrpc_start_thread(dev, svc, name);
 795                 if (rc) {
 796                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
 797                                i, rc);
 798                         ptlrpc_stop_all_threads(svc);
 799                 }
 800         }
 801         RETURN(rc);
 802 }
 803
 804 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
 805                         char *name)
 806 {
 807         struct l_wait_info lwi = { 0 };
 808         struct ptlrpc_svc_data d;
 809         struct ptlrpc_thread *thread;
 810         unsigned long flags;
 811         int rc;
 812         ENTRY;
 813
 814         OBD_ALLOC(thread, sizeof(*thread));
 815         if (thread == NULL)
 816                 RETURN(-ENOMEM);
 817         init_waitqueue_head(&thread->t_ctl_waitq);
 818
 819         d.dev = dev;
 820         d.svc = svc;
 821         d.name = name;
 822         d.thread = thread;
 823
 824         spin_lock_irqsave(&svc->srv_lock, flags);
 825         list_add(&thread->t_link, &svc->srv_threads);
 826         spin_unlock_irqrestore(&svc->srv_lock, flags);
 827
 828         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
 829          * just drop the VM and FILES in ptlrpc_daemonize() right away.
 830          */
 831         rc = kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
 832         if (rc < 0) {
 833                 CERROR("cannot start thread: %d\n", rc);
 834                 OBD_FREE(thread, sizeof(*thread));
 835                 RETURN(rc);
 836         }
 837         l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
 838
 839         RETURN(0);
 840 }
 841 #endif
 842
 843 int ptlrpc_unregister_service(struct ptlrpc_service *service)
 844 {
 845         int                   i;
 846         int                   rc;
 847         unsigned long         flags;
 848         struct ptlrpc_srv_ni *srv_ni;
 849         struct l_wait_info    lwi;
 850         struct list_head     *tmp;
 851
 852         LASSERT(list_empty(&service->srv_threads));
 853
 854         spin_lock (&ptlrpc_all_services_lock);
 855         list_del_init (&service->srv_list);
 856         spin_unlock (&ptlrpc_all_services_lock);
 857
 858         ptlrpc_lprocfs_unregister_service(service);
 859
 860         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 861                 srv_ni = &service->srv_interfaces[i];
 862                 CDEBUG(D_NET, "%s: tearing down interface %s\n",
 863                        service->srv_name, srv_ni->sni_ni->pni_name);
 864
 865                 /* Unlink all the request buffers.  This forces a 'final'
 866                  * event with its 'unlink' flag set for each posted rqbd */
 867                 list_for_each(tmp, &srv_ni->sni_active_rqbds) {
 868                         struct ptlrpc_request_buffer_desc *rqbd =
 869                                 list_entry(tmp, struct ptlrpc_request_buffer_desc,
 870                                            rqbd_list);
 871
 872                         rc = PtlMDUnlink(rqbd->rqbd_md_h);
 873                         LASSERT (rc == PTL_OK || rc == PTL_MD_INVALID);
 874                 }
 875
 876                 /* Wait for the network to release any buffers it's
 877                  * currently filling */
 878                 for (;;) {
 879                         spin_lock_irqsave(&service->srv_lock, flags);
 880                         rc = srv_ni->sni_nrqbd_receiving;
 881                         spin_unlock_irqrestore(&service->srv_lock, flags);
 882
 883                         if (rc == 0)
 884                                 break;
 885
 886                         /* Network access will complete in finite time but
 887                          * the HUGE timeout lets us CWARN for visibility of
 888                          * sluggish NALs */
 889                         lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
 890                         rc = l_wait_event(service->srv_waitq,
 891                                           srv_ni->sni_nrqbd_receiving == 0,
 892                                           &lwi);
 893                         if (rc == -ETIMEDOUT)
 894                                 CWARN("Waiting for request buffers on "
 895                                       "service %s on interface %s ",
 896                                       service->srv_name, srv_ni->sni_ni->pni_name);
 897                 }
 898
 899                 /* schedule all outstanding replies to terminate them */
 900                 spin_lock_irqsave(&service->srv_lock, flags);
 901                 while (!list_empty(&srv_ni->sni_active_replies)) {
 902                         struct ptlrpc_reply_state *rs =
 903                                 list_entry(srv_ni->sni_active_replies.next,
 904                                            struct ptlrpc_reply_state,
 905                                            rs_list);
 906                         ptlrpc_schedule_difficult_reply(rs);
 907                 }
 908                 spin_unlock_irqrestore(&service->srv_lock, flags);
 909         }
 910
 911         /* purge the request queue.  NB No new replies (rqbds all unlinked)
 912          * and no service threads, so I'm the only thread noodling the
 913          * request queue now */
 914         while (!list_empty(&service->srv_request_queue)) {
 915                 struct ptlrpc_request *req =
 916                         list_entry(service->srv_request_queue.next,
 917                                    struct ptlrpc_request,
 918                                    rq_list);
 919
 920                 list_del(&req->rq_list);
 921                 service->srv_n_queued_reqs--;
 922                 service->srv_n_active_reqs++;
 923
 924                 ptlrpc_server_free_request(service, req);
 925         }
 926         LASSERT(service->srv_n_queued_reqs == 0);
 927         LASSERT(service->srv_n_active_reqs == 0);
 928
 929         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 930                 srv_ni = &service->srv_interfaces[i];
 931                 LASSERT(list_empty(&srv_ni->sni_active_rqbds));
 932         }
 933
 934         /* Now free all the request buffers since nothing references them
 935          * any more... */
 936         while (!list_empty(&service->srv_idle_rqbds)) {
 937                 struct ptlrpc_request_buffer_desc *rqbd =
 938                         list_entry(service->srv_idle_rqbds.next,
 939                                    struct ptlrpc_request_buffer_desc,
 940                                    rqbd_list);
 941
 942                 ptlrpc_free_rqbd(rqbd);
 943         }
 944
 945         /* wait for all outstanding replies to complete (they were
 946          * scheduled having been flagged to abort above) */
 947         while (atomic_read(&service->srv_outstanding_replies) != 0) {
 948                 struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
 949
 950                 rc = l_wait_event(service->srv_waitq,
 951                                   !list_empty(&service->srv_reply_queue), &lwi);
 952                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
 953
 954                 if (rc == 0) {
 955                         ptlrpc_server_handle_reply(service);
 956                         continue;
 957                 }
 958                 CWARN("Unexpectedly long timeout %p\n", service);
 959         }
 960
 961         OBD_FREE(service,
 962                  offsetof(struct ptlrpc_service,
 963                           srv_interfaces[ptlrpc_ninterfaces]));
 964         return 0;
 965 }