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