lustre/ptlrpc/events.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 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
  25 #ifdef __KERNEL__
  26 #include <linux/module.h>
  27 #else
  28 #include <liblustre.h>
  29 #endif
  30 #include <linux/obd_class.h>
  31 #include <linux/lustre_net.h>
  32
  33 struct ptlrpc_ni  ptlrpc_interfaces[NAL_MAX_NR];
  34 int               ptlrpc_ninterfaces;
  35
  36 /*
  37  *  Client's outgoing request callback
  38  */
  39 void request_out_callback(ptl_event_t *ev)
  40 {
  41         struct ptlrpc_cb_id   *cbid = ev->mem_desc.user_ptr;
  42         struct ptlrpc_request *req = cbid->cbid_arg;
  43         unsigned long          flags;
  44         ENTRY;
  45
  46         LASSERT (ev->type == PTL_EVENT_SENT ||
  47                  ev->type == PTL_EVENT_UNLINK);
  48         LASSERT (ev->unlinked);
  49
  50         DEBUG_REQ((ev->status == PTL_OK) ? D_NET : D_ERROR, req,
  51                   "type %d, status %d", ev->type, ev->status);
  52
  53         if (ev->type == PTL_EVENT_UNLINK ||
  54             ev->status != PTL_OK) {
  55
  56                 /* Failed send: make it seem like the reply timed out, just
  57                  * like failing sends in client.c does currently...  */
  58
  59                 spin_lock_irqsave(&req->rq_lock, flags);
  60                 req->rq_timeout = 0;
  61                 spin_unlock_irqrestore(&req->rq_lock, flags);
  62
  63                 ptlrpc_wake_client_req(req);
  64         }
  65
  66         /* this balances the atomic_inc in ptl_send_rpc() */
  67         ptlrpc_req_finished(req);
  68         EXIT;
  69 }
  70
  71 /*
  72  * Client's incoming reply callback
  73  */
  74 void reply_in_callback(ptl_event_t *ev)
  75 {
  76         struct ptlrpc_cb_id   *cbid = ev->mem_desc.user_ptr;
  77         struct ptlrpc_request *req = cbid->cbid_arg;
  78         unsigned long flags;
  79         ENTRY;
  80
  81         LASSERT (ev->type == PTL_EVENT_PUT ||
  82                  ev->type == PTL_EVENT_UNLINK);
  83         LASSERT (ev->unlinked);
  84         LASSERT (ev->mem_desc.start == req->rq_repmsg);
  85         LASSERT (ev->offset == 0);
  86         LASSERT (ev->mlength <= req->rq_replen);
  87
  88         DEBUG_REQ((ev->status == PTL_OK) ? D_NET : D_ERROR, req,
  89                   "type %d, status %d", ev->type, ev->status);
  90
  91         spin_lock_irqsave (&req->rq_lock, flags);
  92
  93         LASSERT (req->rq_receiving_reply);
  94         req->rq_receiving_reply = 0;
  95
  96         if (ev->type == PTL_EVENT_PUT &&
  97             ev->status == PTL_OK) {
  98                 req->rq_replied = 1;
  99                 req->rq_nob_received = ev->mlength;
 100         }
 101
 102         /* NB don't unlock till after wakeup; req can disappear under us
 103          * since we don't have our own ref */
 104         ptlrpc_wake_client_req(req);
 105
 106         spin_unlock_irqrestore (&req->rq_lock, flags);
 107         EXIT;
 108 }
 109
 110 /*
 111  * Client's bulk has been written/read
 112  */
 113 void client_bulk_callback (ptl_event_t *ev)
 114 {
 115         struct ptlrpc_cb_id     *cbid = ev->mem_desc.user_ptr;
 116         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 117         unsigned long            flags;
 118         ENTRY;
 119
 120         LASSERT ((desc->bd_type == BULK_PUT_SINK &&
 121                   ev->type == PTL_EVENT_PUT) ||
 122                  (desc->bd_type == BULK_GET_SOURCE &&
 123                   ev->type == PTL_EVENT_GET) ||
 124                  ev->type == PTL_EVENT_UNLINK);
 125         LASSERT (ev->unlinked);
 126
 127         CDEBUG((ev->status == PTL_OK) ? D_NET : D_ERROR,
 128                "event type %d, status %d, desc %p\n",
 129                ev->type, ev->status, desc);
 130
 131         spin_lock_irqsave (&desc->bd_lock, flags);
 132
 133         LASSERT(desc->bd_network_rw);
 134         desc->bd_network_rw = 0;
 135
 136         if (ev->type != PTL_EVENT_UNLINK &&
 137             ev->status == PTL_OK) {
 138                 desc->bd_success = 1;
 139                 desc->bd_nob_transferred = ev->mlength;
 140         }
 141
 142         /* NB don't unlock till after wakeup; desc can disappear under us
 143          * otherwise */
 144         ptlrpc_wake_client_req(desc->bd_req);
 145
 146         spin_unlock_irqrestore (&desc->bd_lock, flags);
 147         EXIT;
 148 }
 149
 150 /*
 151  * Server's incoming request callback
 152  */
 153 void request_in_callback(ptl_event_t *ev)
 154 {
 155         struct ptlrpc_cb_id               *cbid = ev->mem_desc.user_ptr;
 156         struct ptlrpc_request_buffer_desc *rqbd = cbid->cbid_arg;
 157         struct ptlrpc_srv_ni              *srv_ni = rqbd->rqbd_srv_ni;
 158         struct ptlrpc_service             *service = srv_ni->sni_service;
 159         struct ptlrpc_request             *req;
 160         long                               flags;
 161         ENTRY;
 162
 163         LASSERT (ev->type == PTL_EVENT_PUT ||
 164                  ev->type == PTL_EVENT_UNLINK);
 165         LASSERT ((char *)ev->mem_desc.start >= rqbd->rqbd_buffer);
 166         LASSERT ((char *)ev->mem_desc.start + ev->offset + ev->mlength <=
 167                  rqbd->rqbd_buffer + service->srv_buf_size);
 168
 169         CDEBUG((ev->status == PTL_OK) ? D_NET : D_ERROR,
 170                "event type %d, status %d, service %s\n",
 171                ev->type, ev->status, service->srv_name);
 172
 173         if (ev->unlinked) {
 174                 /* If this is the last request message to fit in the
 175                  * request buffer we can use the request object embedded in
 176                  * rqbd.  Note that if we failed to allocate a request,
 177                  * we'd have to re-post the rqbd, which we can't do in this
 178                  * context. */
 179                 req = &rqbd->rqbd_req;
 180                 memset(req, 0, sizeof (*req));
 181         } else {
 182                 LASSERT (ev->type == PTL_EVENT_PUT);
 183                 if (ev->status != PTL_OK) {
 184                         /* We moaned above already... */
 185                         return;
 186                 }
 187                 OBD_ALLOC_GFP(req, sizeof(*req), GFP_ATOMIC);
 188                 if (req == NULL) {
 189                         CERROR("Can't allocate incoming request descriptor: "
 190                                "Dropping %s RPC from "LPX64"\n",
 191                                service->srv_name, ev->initiator.nid);
 192                         return;
 193                 }
 194         }
 195
 196         /* NB we ABSOLUTELY RELY on req being zeroed, so pointers are NULL,
 197          * flags are reset and scalars are zero.  We only set the message
 198          * size to non-zero if this was a successful receive. */
 199         req->rq_xid = ev->match_bits;
 200         req->rq_reqmsg = ev->mem_desc.start + ev->offset;
 201         if (ev->type == PTL_EVENT_PUT &&
 202             ev->status == PTL_OK)
 203                 req->rq_reqlen = ev->mlength;
 204         req->rq_arrival_time = ev->arrival_time;
 205         req->rq_peer.peer_nid = ev->initiator.nid;
 206         req->rq_peer.peer_ni = rqbd->rqbd_srv_ni->sni_ni;
 207         req->rq_rqbd = rqbd;
 208
 209         spin_lock_irqsave (&service->srv_lock, flags);
 210
 211         if (ev->unlinked) {
 212                 srv_ni->sni_nrqbd_receiving--;
 213                 if (ev->type != PTL_EVENT_UNLINK &&
 214                     srv_ni->sni_nrqbd_receiving == 0) {
 215                         /* This service is off-air on this interface because
 216                          * all its request buffers are busy.  Portals will
 217                          * start dropping incoming requests until more buffers
 218                          * get posted.  NB don't moan if it's because we're
 219                          * tearing down the service. */
 220                         CWARN("All %s %s request buffers busy\n",
 221                               service->srv_name, srv_ni->sni_ni->pni_name);
 222                 }
 223                 /* req takes over the network's ref on rqbd */
 224         } else {
 225                 /* req takes a ref on rqbd */
 226                 rqbd->rqbd_refcount++;
 227         }
 228
 229         list_add_tail(&req->rq_list, &service->srv_request_queue);
 230         service->srv_n_queued_reqs++;
 231
 232         /* NB everything can disappear under us once the request
 233          * has been queued and we unlock, so do the wake now... */
 234         wake_up(&service->srv_waitq);
 235
 236         spin_unlock_irqrestore(&service->srv_lock, flags);
 237         EXIT;
 238 }
 239
 240 /*
 241  *  Server's outgoing reply callback
 242  */
 243 void reply_out_callback(ptl_event_t *ev)
 244 {
 245         struct ptlrpc_cb_id       *cbid = ev->mem_desc.user_ptr;
 246         struct ptlrpc_reply_state *rs = cbid->cbid_arg;
 247         struct ptlrpc_srv_ni      *sni = rs->rs_srv_ni;
 248         struct ptlrpc_service     *svc = sni->sni_service;
 249         unsigned long              flags;
 250         ENTRY;
 251
 252         LASSERT (ev->type == PTL_EVENT_SENT ||
 253                  ev->type == PTL_EVENT_ACK ||
 254                  ev->type == PTL_EVENT_UNLINK);
 255
 256         if (!rs->rs_difficult) {
 257                 /* I'm totally responsible for freeing "easy" replies */
 258                 LASSERT (ev->unlinked);
 259                 lustre_free_reply_state (rs);
 260                 atomic_dec (&svc->srv_outstanding_replies);
 261                 EXIT;
 262                 return;
 263         }
 264
 265         LASSERT (rs->rs_on_net);
 266
 267         if (ev->unlinked) {
 268                 /* Last network callback */
 269                 spin_lock_irqsave (&svc->srv_lock, flags);
 270                 rs->rs_on_net = 0;
 271                 ptlrpc_schedule_difficult_reply (rs);
 272                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 273         }
 274
 275         EXIT;
 276 }
 277
 278 /*
 279  * Server's bulk completion callback
 280  */
 281 void server_bulk_callback (ptl_event_t *ev)
 282 {
 283         struct ptlrpc_cb_id     *cbid = ev->mem_desc.user_ptr;
 284         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 285         unsigned long            flags;
 286         ENTRY;
 287
 288         LASSERT (ev->type == PTL_EVENT_SENT ||
 289                  ev->type == PTL_EVENT_UNLINK ||
 290                  (desc->bd_type == BULK_PUT_SOURCE &&
 291                   ev->type == PTL_EVENT_ACK) ||
 292                  (desc->bd_type == BULK_GET_SINK &&
 293                   ev->type == PTL_EVENT_REPLY));
 294
 295         CDEBUG((ev->status == PTL_OK) ? D_NET : D_ERROR,
 296                "event type %d, status %d, desc %p\n",
 297                ev->type, ev->status, desc);
 298
 299         spin_lock_irqsave (&desc->bd_lock, flags);
 300
 301         if ((ev->type == PTL_EVENT_ACK ||
 302              ev->type == PTL_EVENT_REPLY) &&
 303             ev->status == PTL_OK) {
 304                 /* We heard back from the peer, so even if we get this
 305                  * before the SENT event (oh yes we can), we know we
 306                  * read/wrote the peer buffer and how much... */
 307                 desc->bd_success = 1;
 308                 desc->bd_nob_transferred = ev->mlength;
 309         }
 310
 311         if (ev->unlinked) {
 312                 /* This is the last callback no matter what... */
 313                 desc->bd_network_rw = 0;
 314                 wake_up(&desc->bd_waitq);
 315         }
 316
 317         spin_unlock_irqrestore (&desc->bd_lock, flags);
 318         EXIT;
 319 }
 320
 321 static int ptlrpc_master_callback(ptl_event_t *ev)
 322 {
 323         struct ptlrpc_cb_id *cbid = ev->mem_desc.user_ptr;
 324         void (*callback)(ptl_event_t *ev) = cbid->cbid_fn;
 325
 326         /* Honestly, it's best to find out early. */
 327         LASSERT (cbid->cbid_arg != (void *)0x5a5a5a5a5a5a5a5a);
 328         LASSERT (callback == request_out_callback ||
 329                  callback == reply_in_callback ||
 330                  callback == client_bulk_callback ||
 331                  callback == request_in_callback ||
 332                  callback == reply_out_callback ||
 333                  callback == server_bulk_callback);
 334
 335         callback (ev);
 336         return (0);
 337 }
 338
 339 int ptlrpc_uuid_to_peer (struct obd_uuid *uuid, struct ptlrpc_peer *peer)
 340 {
 341         struct ptlrpc_ni   *pni;
 342         struct lustre_peer  lpeer;
 343         int                 i;
 344         int                 rc = lustre_uuid_to_peer (uuid->uuid, &lpeer);
 345
 346         if (rc != 0)
 347                 RETURN (rc);
 348
 349         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 350                 pni = &ptlrpc_interfaces[i];
 351
 352                 if (!memcmp(&lpeer.peer_ni, &pni->pni_ni_h,
 353                             sizeof (lpeer.peer_ni))) {
 354                         peer->peer_nid = lpeer.peer_nid;
 355                         peer->peer_ni = pni;
 356                         return (0);
 357                 }
 358         }
 359
 360         CERROR("Can't find ptlrpc interface for "LPX64" ni handle %08lx."LPX64"\n",
 361                lpeer.peer_nid, lpeer.peer_ni.nal_idx, lpeer.peer_ni.cookie);
 362         return (-ENOENT);
 363 }
 364
 365 void ptlrpc_ni_fini(struct ptlrpc_ni *pni)
 366 {
 367         wait_queue_head_t   waitq;
 368         struct l_wait_info  lwi;
 369         int                 rc;
 370         int                 retries;
 371
 372         /* Wait for the event queue to become idle since there may still be
 373          * messages in flight with pending events (i.e. the fire-and-forget
 374          * messages == client requests and "non-difficult" server
 375          * replies */
 376
 377         for (retries = 0;; retries++) {
 378                 rc = PtlEQFree(pni->pni_eq_h);
 379                 switch (rc) {
 380                 default:
 381                         LBUG();
 382
 383                 case PTL_OK:
 384                         kportal_put_ni (pni->pni_number);
 385                         return;
 386
 387                 case PTL_EQ_INUSE:
 388                         if (retries != 0)
 389                                 CWARN("Event queue for %s still busy\n",
 390                                       pni->pni_name);
 391
 392                         /* Wait for a bit */
 393                         init_waitqueue_head(&waitq);
 394                         lwi = LWI_TIMEOUT(2*HZ, NULL, NULL);
 395                         l_wait_event(waitq, 0, &lwi);
 396                         break;
 397                 }
 398         }
 399         /* notreached */
 400 }
 401
 402 int ptlrpc_ni_init(int number, char *name, struct ptlrpc_ni *pni)
 403 {
 404         int              rc;
 405         ptl_handle_ni_t *nip = kportal_get_ni (number);
 406
 407         if (nip == NULL) {
 408                 CDEBUG (D_NET, "Network interface %s not loaded\n", name);
 409                 return (-ENOENT);
 410         }
 411
 412         CDEBUG (D_NET, "init %d %s: nal_idx %ld\n", number, name, nip->nal_idx);
 413
 414         pni->pni_name = name;
 415         pni->pni_number = number;
 416         pni->pni_ni_h = *nip;
 417
 418         pni->pni_eq_h = PTL_HANDLE_NONE;
 419
 420 #ifdef __KERNEL__
 421         /* kernel: portals calls the callback when the event is added to the
 422          * queue, so we don't care if we lose events */
 423         rc = PtlEQAlloc(pni->pni_ni_h, 1024, ptlrpc_master_callback,
 424                         &pni->pni_eq_h);
 425 #else
 426         /* liblustre: no asynchronous callback and allocate a nice big event
 427          * queue so we don't drop any events... */
 428         rc = PtlEQAlloc(pni->pni_ni_h, 10240, NULL, &pni->pni_eq_h);
 429 #endif
 430         if (rc != PTL_OK)
 431                 GOTO (fail, rc = -ENOMEM);
 432
 433         return (0);
 434  fail:
 435         CERROR ("Failed to initialise network interface %s: %d\n",
 436                 name, rc);
 437
 438         /* OK to do complete teardown since we invalidated the handles above */
 439         ptlrpc_ni_fini (pni);
 440         return (rc);
 441 }
 442
 443 #ifndef __KERNEL__
 444 LIST_HEAD(liblustre_wait_callbacks);
 445 void *liblustre_services_callback;
 446
 447 void *
 448 liblustre_register_wait_callback (int (*fn)(void *arg), void *arg)
 449 {
 450         struct liblustre_wait_callback *llwc;
 451
 452         OBD_ALLOC(llwc, sizeof(*llwc));
 453         LASSERT (llwc != NULL);
 454
 455         llwc->llwc_fn = fn;
 456         llwc->llwc_arg = arg;
 457         list_add_tail(&llwc->llwc_list, &liblustre_wait_callbacks);
 458
 459         return (llwc);
 460 }
 461
 462 void
 463 liblustre_deregister_wait_callback (void *opaque)
 464 {
 465         struct liblustre_wait_callback *llwc = opaque;
 466
 467         list_del(&llwc->llwc_list);
 468         OBD_FREE(llwc, sizeof(*llwc));
 469 }
 470
 471 int
 472 liblustre_check_events (int timeout)
 473 {
 474         ptl_event_t ev;
 475         int         rc;
 476         ENTRY;
 477
 478         if (timeout) {
 479                 rc = PtlEQWait_timeout(ptlrpc_interfaces[0].pni_eq_h, &ev, timeout);
 480         } else {
 481                 rc = PtlEQGet (ptlrpc_interfaces[0].pni_eq_h, &ev);
 482         }
 483         if (rc == PTL_EQ_EMPTY)
 484                 RETURN(0);
 485
 486         LASSERT (rc == PTL_EQ_DROPPED || rc == PTL_OK);
 487
 488 #ifndef __KERNEL__
 489         /* liblustre: no asynch callback so we can't affort to miss any
 490          * events... */
 491         if (rc == PTL_EQ_DROPPED) {
 492                 CERROR ("Dropped an event!!!\n");
 493                 abort();
 494         }
 495
 496         ptlrpc_master_callback (&ev);
 497 #endif
 498         RETURN(1);
 499 }
 500
 501 int
 502 liblustre_wait_event (int timeout)
 503 {
 504         struct list_head               *tmp;
 505         struct liblustre_wait_callback *llwc;
 506         int                             found_something = 0;
 507
 508         /* First check for any new events */
 509         if (liblustre_check_events(0))
 510                 found_something = 1;
 511
 512         /* Now give all registered callbacks a bite at the cherry */
 513         list_for_each(tmp, &liblustre_wait_callbacks) {
 514                 llwc = list_entry(tmp, struct liblustre_wait_callback,
 515                                   llwc_list);
 516
 517                 if (llwc->llwc_fn(llwc->llwc_arg))
 518                         found_something = 1;
 519         }
 520
 521         /* return to caller if something happened */
 522         if (found_something)
 523                 return 1;
 524
 525         /* block for an event, returning immediately on timeout */
 526         if (!liblustre_check_events(timeout))
 527                 return 0;
 528
 529         /* an event occurred; let all registered callbacks progress... */
 530         list_for_each(tmp, &liblustre_wait_callbacks) {
 531                 llwc = list_entry(tmp, struct liblustre_wait_callback,
 532                                   llwc_list);
 533
 534                 if (llwc->llwc_fn(llwc->llwc_arg))
 535                         found_something = 1;
 536         }
 537
 538         /* ...and tell caller something happened */
 539         return 1;
 540 }
 541 #endif
 542
 543 int ptlrpc_init_portals(void)
 544 {
 545         /* Add new portals network interfaces here.
 546          * Order is irrelevent! */
 547         static struct {
 548                 int   number;
 549                 char *name;
 550         } ptl_nis[] = {
 551                 {QSWNAL,  "qswnal"},
 552                 {SOCKNAL, "socknal"},
 553                 {GMNAL,   "gmnal"},
 554                 {IBNAL,   "ibnal"},
 555                 {TCPNAL,  "tcpnal"},
 556                 {SCIMACNAL, "scimacnal"}};
 557         int   rc;
 558         int   i;
 559
 560         LASSERT(ptlrpc_ninterfaces == 0);
 561
 562         for (i = 0; i < sizeof (ptl_nis) / sizeof (ptl_nis[0]); i++) {
 563                 LASSERT(ptlrpc_ninterfaces < (sizeof(ptlrpc_interfaces) /
 564                                               sizeof(ptlrpc_interfaces[0])));
 565
 566                 rc = ptlrpc_ni_init(ptl_nis[i].number, ptl_nis[i].name,
 567                                     &ptlrpc_interfaces[ptlrpc_ninterfaces]);
 568                 if (rc == 0)
 569                         ptlrpc_ninterfaces++;
 570         }
 571
 572         if (ptlrpc_ninterfaces == 0) {
 573                 CERROR("network initialisation failed: is a NAL module "
 574                        "loaded?\n");
 575                 return -EIO;
 576         }
 577 #ifndef __KERNEL__
 578         liblustre_services_callback =
 579                 liblustre_register_wait_callback(&liblustre_check_services, NULL);
 580 #endif
 581         return 0;
 582 }
 583
 584 void ptlrpc_exit_portals(void)
 585 {
 586 #ifndef __KERNEL__
 587         liblustre_deregister_wait_callback(liblustre_services_callback);
 588 #endif
 589         while (ptlrpc_ninterfaces > 0)
 590                 ptlrpc_ni_fini (&ptlrpc_interfaces[--ptlrpc_ninterfaces]);
 591 }