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