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