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