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         request->rq_mapped_uid = -1;
 463         request->rq_remote_realm = 0;
 464
 465         secrc = svcsec_accept(request, &sec_err);
 466         switch(secrc) {
 467         case SVC_OK:
 468                 CDEBUG(D_SEC, "request accepted ok\n");
 469                 break;
 470         case SVC_COMPLETE:
 471                 target_send_reply(request, 0, OBD_FAIL_MDS_ALL_REPLY_NET);
 472                 goto put_conn;
 473         case SVC_LOGIN:
 474         case SVC_LOGOUT:
 475                 break;
 476         case SVC_DROP:
 477                 goto out;
 478         default:
 479                 LBUG();
 480         }
 481
 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         /* Discard requests queued for longer than my timeout.  If the
 500          * client's timeout is similar to mine, she'll be timing out this
 501          * REQ anyway (bug 1502) */
 502         if (timediff / 1000000 > (long)obd_timeout) {
 503                 CERROR("Dropping timed-out opc %d request from %s"
 504                        ": %ld seconds old\n", request->rq_reqmsg->opc,
 505                        request->rq_peerstr,
 506                        timediff / 1000000);
 507                 goto out;
 508         }
 509
 510         request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
 511
 512         if (request->rq_export) {
 513                 if (request->rq_reqmsg->conn_cnt <
 514                     request->rq_export->exp_conn_cnt) {
 515                         DEBUG_REQ(D_ERROR, request,
 516                                   "DROPPING req from old connection %d < %d",
 517                                   request->rq_reqmsg->conn_cnt,
 518                                   request->rq_export->exp_conn_cnt);
 519                         goto put_conn;
 520                 }
 521
 522                 export = class_export_rpc_get(request->rq_export);
 523                 request->rq_export->exp_last_request_time =
 524                         LTIME_S(CURRENT_TIME);
 525         }
 526
 527         CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 528                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
 529                (request->rq_export ?
 530                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 531                (request->rq_export ?
 532                 atomic_read(&request->rq_export->exp_refcount) : -99),
 533                request->rq_reqmsg->status, request->rq_xid,
 534                request->rq_peer.peer_ni->pni_name,
 535                request->rq_peerstr,
 536                request->rq_reqmsg->opc);
 537
 538         request->rq_svc = svc;
 539         rc = svc->srv_handler(request);
 540         request->rq_svc = NULL;
 541
 542         CDEBUG(D_RPCTRACE, "Handled RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
 543                "%s:%s+%d:%d:"LPU64":%s:%s:%d\n", current->comm,
 544                (request->rq_export ?
 545                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
 546                (request->rq_export ?
 547                 atomic_read(&request->rq_export->exp_refcount) : -99),
 548                request->rq_reqmsg->status, request->rq_xid,
 549                request->rq_peer.peer_ni->pni_name,
 550                request->rq_peerstr,
 551                request->rq_reqmsg->opc);
 552
 553         if (export != NULL)
 554                 class_export_rpc_put(export);
 555
 556 put_conn:
 557         if (request->rq_export != NULL)
 558                 class_export_put(request->rq_export);
 559
 560  out:
 561         do_gettimeofday(&work_end);
 562
 563         timediff = timeval_sub(&work_end, &work_start);
 564
 565         CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
 566                "request "LPU64" opc %u from NID %s processed in %ldus "
 567                "(%ldus total)\n", request->rq_xid,
 568                request->rq_reqmsg ? request->rq_reqmsg->opc : 0,
 569                request->rq_peerstr,
 570                timediff, timeval_sub(&work_end, &request->rq_arrival_time));
 571
 572         if (svc->srv_stats != NULL && request->rq_reqmsg != NULL) {
 573                 int opc = opcode_offset(request->rq_reqmsg->opc);
 574                 if (opc > 0) {
 575                         LASSERT(opc < LUSTRE_MAX_OPCODES);
 576                         lprocfs_counter_add(svc->srv_stats,
 577                                             opc + PTLRPC_LAST_CNTR,
 578                                             timediff);
 579                 }
 580         }
 581
 582         ptlrpc_server_free_request(svc, request);
 583
 584         RETURN(1);
 585 }
 586
 587 static int
 588 ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
 589 {
 590         struct ptlrpc_reply_state *rs;
 591         unsigned long              flags;
 592         struct obd_export         *exp;
 593         struct obd_device         *obd;
 594         struct llog_create_locks  *lcl;
 595         int                        nlocks;
 596         int                        been_handled;
 597         ENTRY;
 598
 599         spin_lock_irqsave (&svc->srv_lock, flags);
 600         if (list_empty (&svc->srv_reply_queue)) {
 601                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 602                 RETURN(0);
 603         }
 604
 605         rs = list_entry (svc->srv_reply_queue.next,
 606                          struct ptlrpc_reply_state, rs_list);
 607
 608         exp = rs->rs_export;
 609         obd = exp->exp_obd;
 610
 611         LASSERT (rs->rs_difficult);
 612         LASSERT (rs->rs_scheduled);
 613
 614         list_del_init (&rs->rs_list);
 615
 616         /* Disengage from notifiers carefully (lock ordering!) */
 617         spin_unlock(&svc->srv_lock);
 618
 619         spin_lock (&obd->obd_uncommitted_replies_lock);
 620         /* Noop if removed already */
 621         list_del_init (&rs->rs_obd_list);
 622         spin_unlock (&obd->obd_uncommitted_replies_lock);
 623
 624         spin_lock (&exp->exp_lock);
 625         /* Noop if removed already */
 626         list_del_init (&rs->rs_exp_list);
 627         spin_unlock (&exp->exp_lock);
 628
 629         spin_lock(&svc->srv_lock);
 630
 631         been_handled = rs->rs_handled;
 632         rs->rs_handled = 1;
 633
 634         nlocks = rs->rs_nlocks;                 /* atomic "steal", but */
 635         rs->rs_nlocks = 0;                      /* locks still on rs_locks! */
 636
 637         lcl = rs->rs_llog_locks;
 638         rs->rs_llog_locks = NULL;
 639
 640         if (nlocks == 0 && !been_handled) {
 641                 /* If we see this, we should already have seen the warning
 642                  * in mds_steal_ack_locks()  */
 643 #if 0
 644                 char str[PTL_NALFMT_SIZE];
 645                 /* CMD may ask to save request with no DLM locks -bzzz */
 646                 CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
 647                       " o%d NID %s\n",
 648                       rs,
 649                       rs->rs_xid, rs->rs_transno,
 650                       rs->rs_msg->opc,
 651                       ptlrpc_peernid2str(&exp->exp_connection->c_peer, str));
 652 #endif
 653         }
 654
 655         if ((!been_handled && rs->rs_on_net) ||
 656             nlocks > 0 || lcl != NULL) {
 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                 if (lcl != NULL)
 670                         llog_create_lock_free(lcl);
 671
 672                 spin_lock_irqsave(&svc->srv_lock, flags);
 673         }
 674
 675         rs->rs_scheduled = 0;
 676
 677         if (!rs->rs_on_net) {
 678                 /* Off the net */
 679                 svc->srv_n_difficult_replies--;
 680                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 681
 682                 class_export_put (exp);
 683                 rs->rs_export = NULL;
 684                 lustre_free_reply_state (rs);
 685                 atomic_dec (&svc->srv_outstanding_replies);
 686                 RETURN(1);
 687         }
 688
 689         /* still on the net; callback will schedule */
 690         spin_unlock_irqrestore (&svc->srv_lock, flags);
 691         RETURN(1);
 692 }
 693
 694 #ifndef __KERNEL__
 695 /* FIXME make use of timeout later */
 696 int
 697 liblustre_check_services (void *arg)
 698 {
 699         int  did_something = 0;
 700         int  rc;
 701         struct list_head *tmp, *nxt;
 702         ENTRY;
 703
 704         /* I'm relying on being single threaded, not to have to lock
 705          * ptlrpc_all_services etc */
 706         list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
 707                 struct ptlrpc_service *svc =
 708                         list_entry (tmp, struct ptlrpc_service, srv_list);
 709
 710                 if (svc->srv_nthreads != 0)     /* I've recursed */
 711                         continue;
 712
 713                 /* service threads can block for bulk, so this limits us
 714                  * (arbitrarily) to recursing 1 stack frame per service.
 715                  * Note that the problem with recursion is that we have to
 716                  * unwind completely before our caller can resume. */
 717
 718                 svc->srv_nthreads++;
 719
 720                 do {
 721                         rc = ptlrpc_server_handle_reply(svc);
 722                         rc |= ptlrpc_server_handle_request(svc);
 723                         rc |= (ptlrpc_server_post_idle_rqbds(svc) > 0);
 724                         did_something |= rc;
 725                 } while (rc);
 726
 727                 svc->srv_nthreads--;
 728         }
 729
 730         RETURN(did_something);
 731 }
 732
 733 #else /* __KERNEL__ */
 734
 735 /* Don't use daemonize, it removes fs struct from new thread (bug 418) */
 736 void ptlrpc_daemonize(void)
 737 {
 738         exit_mm(current);
 739         lustre_daemonize_helper();
 740         exit_files(current);
 741         reparent_to_init();
 742 }
 743
 744 static void
 745 ptlrpc_check_rqbd_pools(struct ptlrpc_service *svc)
 746 {
 747         struct ptlrpc_srv_ni  *sni;
 748         int                    i;
 749         int                    avail = 0;
 750         int                    low_water = svc->srv_nbuf_per_group/2;
 751
 752         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 753                 sni = &svc->srv_interfaces[i];
 754
 755                 avail += sni->sni_nrqbd_receiving;
 756                 /* NB I'm not locking; just looking. */
 757                 if (sni->sni_nrqbd_receiving <= low_water)
 758                         ptlrpc_grow_req_bufs(sni);
 759         }
 760
 761         lprocfs_counter_add(svc->srv_stats, PTLRPC_REQBUF_AVAIL_CNTR, avail);
 762 }
 763
 764 static int
 765 ptlrpc_retry_rqbds(void *arg)
 766 {
 767         struct ptlrpc_service *svc = (struct ptlrpc_service *)arg;
 768
 769         svc->srv_rqbd_timeout = 0;
 770         return (-ETIMEDOUT);
 771 }
 772
 773 static int ptlrpc_main(void *arg)
 774 {
 775         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
 776         struct ptlrpc_service  *svc = data->svc;
 777         struct ptlrpc_thread   *thread = data->thread;
 778         struct lc_watchdog     *watchdog;
 779         unsigned long           flags;
 780 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,4)
 781         struct group_info *ginfo = NULL;
 782 #endif
 783         ENTRY;
 784
 785         lock_kernel();
 786         ptlrpc_daemonize();
 787
 788         SIGNAL_MASK_LOCK(current, flags);
 789         sigfillset(&current->blocked);
 790         RECALC_SIGPENDING;
 791         SIGNAL_MASK_UNLOCK(current, flags);
 792
 793         LASSERTF(strlen(data->name) < sizeof(current->comm),
 794                  "name %d > len %d\n",
 795                  (int)strlen(data->name), (int)sizeof(current->comm));
 796         THREAD_NAME(current->comm, sizeof(current->comm) - 1, "%s", data->name);
 797
 798         unlock_kernel();
 799
 800 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,4)
 801         ginfo = groups_alloc(0);
 802         if (!ginfo) {
 803                 thread->t_flags = SVC_RUNNING;
 804                 wake_up(&thread->t_ctl_waitq);
 805                 return (-ENOMEM);
 806         }
 807         set_current_groups(ginfo);
 808         put_group_info(ginfo);
 809 #endif
 810
 811         /* Record that the thread is running */
 812         thread->t_flags = SVC_RUNNING;
 813         wake_up(&thread->t_ctl_waitq);
 814
 815         watchdog = lc_watchdog_add(svc->srv_watchdog_timeout,
 816                                    LC_WATCHDOG_DEFAULT_CB, NULL);
 817
 818         spin_lock_irqsave(&svc->srv_lock, flags);
 819         svc->srv_nthreads++;
 820         spin_unlock_irqrestore(&svc->srv_lock, flags);
 821
 822         /* XXX maintain a list of all managed devices: insert here */
 823
 824         while ((thread->t_flags & SVC_STOPPING) == 0 ||
 825                svc->srv_n_difficult_replies != 0) {
 826                 /* Don't exit while there are replies to be handled */
 827                 struct l_wait_info lwi = LWI_TIMEOUT(svc->srv_rqbd_timeout,
 828                                                      ptlrpc_retry_rqbds, svc);
 829
 830                 lc_watchdog_disable(watchdog);
 831
 832                 l_wait_event_exclusive (svc->srv_waitq,
 833                               ((thread->t_flags & SVC_STOPPING) != 0 &&
 834                                svc->srv_n_difficult_replies == 0) ||
 835                               (!list_empty(&svc->srv_idle_rqbds) &&
 836                                svc->srv_rqbd_timeout == 0) ||
 837                               !list_empty (&svc->srv_reply_queue) ||
 838                               (!list_empty (&svc->srv_request_queue) &&
 839                                (svc->srv_n_difficult_replies == 0 ||
 840                                 svc->srv_n_active_reqs <
 841                                 (svc->srv_nthreads - 1))),
 842                               &lwi);
 843
 844                 lc_watchdog_touch(watchdog);
 845                 ptlrpc_check_rqbd_pools(svc);
 846
 847                 if (!list_empty (&svc->srv_reply_queue))
 848                         ptlrpc_server_handle_reply (svc);
 849
 850                 /* only handle requests if there are no difficult replies
 851                  * outstanding, or I'm not the last thread handling
 852                  * requests */
 853                 if (!list_empty (&svc->srv_request_queue) &&
 854                     (svc->srv_n_difficult_replies == 0 ||
 855                      svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
 856                         ptlrpc_server_handle_request (svc);
 857
 858                 if (!list_empty(&svc->srv_idle_rqbds) &&
 859                     ptlrpc_server_post_idle_rqbds(svc) < 0) {
 860                         /* I just failed to repost request buffers.  Wait
 861                          * for a timeout (unless something else happens)
 862                          * before I try again */
 863                         svc->srv_rqbd_timeout = HZ/10;
 864                 }
 865         }
 866
 867         spin_lock_irqsave(&svc->srv_lock, flags);
 868
 869         svc->srv_nthreads--;                    /* must know immediately */
 870         thread->t_flags = SVC_STOPPED;
 871         wake_up(&thread->t_ctl_waitq);
 872
 873         spin_unlock_irqrestore(&svc->srv_lock, flags);
 874
 875         lc_watchdog_delete(watchdog);
 876
 877         CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
 878         return 0;
 879 }
 880
 881 static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
 882                                struct ptlrpc_thread *thread)
 883 {
 884         struct l_wait_info lwi = { 0 };
 885         unsigned long      flags;
 886
 887         spin_lock_irqsave(&svc->srv_lock, flags);
 888         thread->t_flags = SVC_STOPPING;
 889         spin_unlock_irqrestore(&svc->srv_lock, flags);
 890
 891         wake_up_all(&svc->srv_waitq);
 892         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
 893                      &lwi);
 894
 895         spin_lock_irqsave(&svc->srv_lock, flags);
 896         list_del(&thread->t_link);
 897         spin_unlock_irqrestore(&svc->srv_lock, flags);
 898
 899         OBD_FREE(thread, sizeof(*thread));
 900 }
 901
 902 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
 903 {
 904         unsigned long flags;
 905         struct ptlrpc_thread *thread;
 906
 907         spin_lock_irqsave(&svc->srv_lock, flags);
 908         while (!list_empty(&svc->srv_threads)) {
 909                 thread = list_entry(svc->srv_threads.next,
 910                                     struct ptlrpc_thread, t_link);
 911
 912                 spin_unlock_irqrestore(&svc->srv_lock, flags);
 913                 ptlrpc_stop_thread(svc, thread);
 914                 spin_lock_irqsave(&svc->srv_lock, flags);
 915         }
 916
 917         spin_unlock_irqrestore(&svc->srv_lock, flags);
 918 }
 919
 920 /* @base_name should be 12 characters or less - 3 will be added on */
 921 int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
 922                            int num_threads, char *base_name)
 923 {
 924         int i, rc = 0;
 925         ENTRY;
 926
 927         for (i = 0; i < num_threads; i++) {
 928                 char name[32];
 929                 sprintf(name, "%s_%02d", base_name, i);
 930                 rc = ptlrpc_start_thread(dev, svc, name);
 931                 if (rc) {
 932                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
 933                                i, rc);
 934                         ptlrpc_stop_all_threads(svc);
 935                 }
 936         }
 937         RETURN(rc);
 938 }
 939
 940 int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
 941                         char *name)
 942 {
 943         struct l_wait_info lwi = { 0 };
 944         struct ptlrpc_svc_data d;
 945         struct ptlrpc_thread *thread;
 946         unsigned long flags;
 947         int rc;
 948         ENTRY;
 949
 950         OBD_ALLOC(thread, sizeof(*thread));
 951         if (thread == NULL)
 952                 RETURN(-ENOMEM);
 953         init_waitqueue_head(&thread->t_ctl_waitq);
 954
 955         d.dev = dev;
 956         d.svc = svc;
 957         d.name = name;
 958         d.thread = thread;
 959
 960         spin_lock_irqsave(&svc->srv_lock, flags);
 961         list_add(&thread->t_link, &svc->srv_threads);
 962         spin_unlock_irqrestore(&svc->srv_lock, flags);
 963
 964         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
 965          * just drop the VM and FILES in ptlrpc_daemonize() right away.
 966          */
 967         rc = kernel_thread(ptlrpc_main, &d, CLONE_VM | CLONE_FILES);
 968         if (rc < 0) {
 969                 CERROR("cannot start thread: %d\n", rc);
 970                 OBD_FREE(thread, sizeof(*thread));
 971                 RETURN(rc);
 972         }
 973         l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
 974
 975         RETURN(0);
 976 }
 977 #endif
 978
 979 int ptlrpc_unregister_service(struct ptlrpc_service *service)
 980 {
 981         int                   i;
 982         int                   rc;
 983         unsigned long         flags;
 984         struct ptlrpc_srv_ni *srv_ni;
 985         struct l_wait_info    lwi;
 986         struct list_head     *tmp;
 987
 988         LASSERT(list_empty(&service->srv_threads));
 989
 990         spin_lock (&ptlrpc_all_services_lock);
 991         list_del_init (&service->srv_list);
 992         spin_unlock (&ptlrpc_all_services_lock);
 993
 994         ptlrpc_lprocfs_unregister_service(service);
 995
 996         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 997                 srv_ni = &service->srv_interfaces[i];
 998                 CDEBUG(D_NET, "%s: tearing down interface %s\n",
 999                        service->srv_name, srv_ni->sni_ni->pni_name);
1000
1001                 /* Unlink all the request buffers.  This forces a 'final'
1002                  * event with its 'unlink' flag set for each posted rqbd */
1003                 list_for_each(tmp, &srv_ni->sni_active_rqbds) {
1004                         struct ptlrpc_request_buffer_desc *rqbd =
1005                                 list_entry(tmp, struct ptlrpc_request_buffer_desc,
1006                                            rqbd_list);
1007
1008                         rc = PtlMDUnlink(rqbd->rqbd_md_h);
1009                         LASSERT (rc == PTL_OK || rc == PTL_MD_INVALID);
1010                 }
1011
1012                 /* Wait for the network to release any buffers it's
1013                  * currently filling */
1014                 for (;;) {
1015                         spin_lock_irqsave(&service->srv_lock, flags);
1016                         rc = srv_ni->sni_nrqbd_receiving;
1017                         spin_unlock_irqrestore(&service->srv_lock, flags);
1018
1019                         if (rc == 0)
1020                                 break;
1021
1022                         /* Network access will complete in finite time but
1023                          * the HUGE timeout lets us CWARN for visibility of
1024                          * sluggish NALs */
1025                         lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
1026                         rc = l_wait_event(service->srv_waitq,
1027                                           srv_ni->sni_nrqbd_receiving == 0,
1028                                           &lwi);
1029                         if (rc == -ETIMEDOUT)
1030                                 CWARN("Waiting for request buffers on "
1031                                       "service %s on interface %s ",
1032                                       service->srv_name, srv_ni->sni_ni->pni_name);
1033                 }
1034
1035                 /* schedule all outstanding replies to terminate them */
1036                 spin_lock_irqsave(&service->srv_lock, flags);
1037                 while (!list_empty(&srv_ni->sni_active_replies)) {
1038                         struct ptlrpc_reply_state *rs =
1039                                 list_entry(srv_ni->sni_active_replies.next,
1040                                            struct ptlrpc_reply_state,
1041                                            rs_list);
1042                         ptlrpc_schedule_difficult_reply(rs);
1043                 }
1044                 spin_unlock_irqrestore(&service->srv_lock, flags);
1045         }
1046
1047         /* purge the request queue.  NB No new replies (rqbds all unlinked)
1048          * and no service threads, so I'm the only thread noodling the
1049          * request queue now */
1050         while (!list_empty(&service->srv_request_queue)) {
1051                 struct ptlrpc_request *req =
1052                         list_entry(service->srv_request_queue.next,
1053                                    struct ptlrpc_request,
1054                                    rq_list);
1055
1056                 list_del(&req->rq_list);
1057                 service->srv_n_queued_reqs--;
1058                 service->srv_n_active_reqs++;
1059
1060                 ptlrpc_server_free_request(service, req);
1061         }
1062         LASSERT(service->srv_n_queued_reqs == 0);
1063         LASSERT(service->srv_n_active_reqs == 0);
1064
1065         for (i = 0; i < ptlrpc_ninterfaces; i++) {
1066                 srv_ni = &service->srv_interfaces[i];
1067                 LASSERT(list_empty(&srv_ni->sni_active_rqbds));
1068         }
1069
1070         /* Now free all the request buffers since nothing references them
1071          * any more... */
1072         while (!list_empty(&service->srv_idle_rqbds)) {
1073                 struct ptlrpc_request_buffer_desc *rqbd =
1074                         list_entry(service->srv_idle_rqbds.next,
1075                                    struct ptlrpc_request_buffer_desc,
1076                                    rqbd_list);
1077
1078                 ptlrpc_free_rqbd(rqbd);
1079         }
1080
1081         /* wait for all outstanding replies to complete (they were
1082          * scheduled having been flagged to abort above) */
1083         while (atomic_read(&service->srv_outstanding_replies) != 0) {
1084                 struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
1085
1086                 rc = l_wait_event(service->srv_waitq,
1087                                   !list_empty(&service->srv_reply_queue), &lwi);
1088                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
1089
1090                 if (rc == 0) {
1091                         ptlrpc_server_handle_reply(service);
1092                         continue;
1093                 }
1094                 CWARN("Unexpectedly long timeout %p\n", service);
1095         }
1096
1097         OBD_FREE(service,
1098                  offsetof(struct ptlrpc_service,
1099                           srv_interfaces[ptlrpc_ninterfaces]));
1100         return 0;
1101 }