Whamcloud - gitweb
* Print portals error string in ptllnd warnings/errors
[fs/lustre-release.git] / lnet / ulnds / ptllnd / ptllnd_cb.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * Copyright (C) 2005 Cluster File Systems, Inc. All rights reserved.
5  *   Author: Eric Barton <eeb@bartonsoftware.com>
6  *
7  *   This file is part of the Lustre file system, http://www.lustre.org
8  *   Lustre is a trademark of Cluster File Systems, Inc.
9  *
10  *   This file is confidential source code owned by Cluster File Systems.
11  *   No viewing, modification, compilation, redistribution, or any other
12  *   form of use is permitted except through a signed license agreement.
13  *
14  *   If you have not signed such an agreement, then you have no rights to
15  *   this file.  Please destroy it immediately and contact CFS.
16  *
17  */
18
19 #include "ptllnd.h"
20
21 void
22 ptllnd_set_tx_deadline(ptllnd_tx_t *tx)
23 {
24         ptllnd_peer_t  *peer = tx->tx_peer;
25         lnet_ni_t      *ni = peer->plp_ni;
26         ptllnd_ni_t    *plni = ni->ni_data;
27
28         tx->tx_deadline = cfs_time_current_sec() + plni->plni_timeout;
29 }
30
31 void
32 ptllnd_post_tx(ptllnd_tx_t *tx)
33 {
34         ptllnd_peer_t  *peer = tx->tx_peer;
35
36         ptllnd_set_tx_deadline(tx);
37         list_add_tail(&tx->tx_list, &peer->plp_txq);
38         ptllnd_check_sends(peer);
39 }
40
41 char *
42 ptllnd_ptlid2str(ptl_process_id_t id)
43 {
44         static char strs[8][32];
45         static int  idx = 0;
46
47         char   *str = strs[idx++];
48         
49         if (idx >= sizeof(strs)/sizeof(strs[0]))
50                 idx = 0;
51
52         snprintf(str, sizeof(strs[0]), FMT_PTLID, id.pid, id.nid);
53         return str;
54 }
55
56 void
57 ptllnd_destroy_peer(ptllnd_peer_t *peer)
58 {
59         lnet_ni_t         *ni = peer->plp_ni;
60         ptllnd_ni_t       *plni = ni->ni_data;
61         int                nmsg = peer->plp_lazy_credits +
62                                   plni->plni_peer_credits;
63
64         ptllnd_size_buffers(ni, -nmsg);
65
66         LASSERT (peer->plp_closing);
67         LASSERT (plni->plni_npeers > 0);
68         LASSERT (list_empty(&peer->plp_txq));
69         LASSERT (list_empty(&peer->plp_activeq));
70         plni->plni_npeers--;
71         LIBCFS_FREE(peer, sizeof(*peer));
72 }
73
74 void
75 ptllnd_abort_txs(ptllnd_ni_t *plni, struct list_head *q)
76 {
77         while (!list_empty(q)) {
78                 ptllnd_tx_t *tx = list_entry(q->next, ptllnd_tx_t, tx_list);
79
80                 tx->tx_status = -ESHUTDOWN;
81                 list_del(&tx->tx_list);
82                 list_add_tail(&tx->tx_list, &plni->plni_zombie_txs);
83         }
84 }
85
86 void
87 ptllnd_close_peer(ptllnd_peer_t *peer, int error)
88 {
89         lnet_ni_t   *ni = peer->plp_ni;
90         ptllnd_ni_t *plni = ni->ni_data;
91
92         if (peer->plp_closing)
93                 return;
94
95         peer->plp_closing = 1;
96
97         if (!list_empty(&peer->plp_txq) ||
98             !list_empty(&peer->plp_activeq) ||
99             error != 0) {
100                 CWARN("Closing %s\n", libcfs_id2str(peer->plp_id));
101                 if (plni->plni_debug)
102                         ptllnd_dump_debug(ni, peer->plp_id);
103         }
104         
105         ptllnd_abort_txs(plni, &peer->plp_txq);
106         ptllnd_abort_txs(plni, &peer->plp_activeq);
107
108         list_del(&peer->plp_list);
109         ptllnd_peer_decref(peer);
110 }
111
112 ptllnd_peer_t *
113 ptllnd_find_peer(lnet_ni_t *ni, lnet_process_id_t id, int create)
114 {
115         ptllnd_ni_t       *plni = ni->ni_data;
116         unsigned int       hash = LNET_NIDADDR(id.nid) % plni->plni_peer_hash_size;
117         struct list_head  *tmp;
118         ptllnd_peer_t     *plp;
119         ptllnd_tx_t       *tx;
120         int                rc;
121
122         LASSERT (LNET_NIDNET(id.nid) == LNET_NIDNET(ni->ni_nid));
123
124         list_for_each(tmp, &plni->plni_peer_hash[hash]) {
125                 plp = list_entry(tmp, ptllnd_peer_t, plp_list);
126
127                 if (plp->plp_id.nid == id.nid &&
128                     plp->plp_id.pid == id.pid) {
129                         ptllnd_peer_addref(plp);
130                         return plp;
131                 }
132         }
133
134         if (!create)
135                 return NULL;
136
137         /* New peer: check first for enough posted buffers */
138         plni->plni_npeers++;
139         rc = ptllnd_size_buffers(ni, plni->plni_peer_credits);
140         if (rc != 0) {
141                 plni->plni_npeers--;
142                 return NULL;
143         }
144
145         LIBCFS_ALLOC(plp, sizeof(*plp));
146         if (plp == NULL) {
147                 CERROR("Can't allocate new peer %s\n", libcfs_id2str(id));
148                 plni->plni_npeers--;
149                 ptllnd_size_buffers(ni, -plni->plni_peer_credits);
150                 return NULL;
151         }
152
153         plp->plp_ni = ni;
154         plp->plp_id = id;
155         plp->plp_ptlid.nid = LNET_NIDADDR(id.nid);
156         plp->plp_ptlid.pid = plni->plni_ptllnd_pid;
157         plp->plp_credits = 1; /* add more later when she gives me credits */
158         plp->plp_max_msg_size = plni->plni_max_msg_size; /* until I hear from her */
159         plp->plp_sent_credits = 1;              /* Implicit credit for HELLO */
160         plp->plp_outstanding_credits = plni->plni_peer_credits - 1;
161         plp->plp_lazy_credits = 0;
162         plp->plp_extra_lazy_credits = 0;
163         plp->plp_match = 0;
164         plp->plp_stamp = 0;
165         plp->plp_recvd_hello = 0;
166         plp->plp_closing = 0;
167         plp->plp_refcount = 1;
168         CFS_INIT_LIST_HEAD(&plp->plp_list);
169         CFS_INIT_LIST_HEAD(&plp->plp_txq);
170         CFS_INIT_LIST_HEAD(&plp->plp_activeq);
171
172         ptllnd_peer_addref(plp);
173         list_add_tail(&plp->plp_list, &plni->plni_peer_hash[hash]);
174
175         tx = ptllnd_new_tx(plp, PTLLND_MSG_TYPE_HELLO, 0);
176         if (tx == NULL) {
177                 CERROR("Can't send HELLO to %s\n", libcfs_id2str(id));
178                 ptllnd_close_peer(plp, -ENOMEM);
179                 ptllnd_peer_decref(plp);
180                 return NULL;
181         }
182
183         tx->tx_msg.ptlm_u.hello.kptlhm_matchbits = PTL_RESERVED_MATCHBITS;
184         tx->tx_msg.ptlm_u.hello.kptlhm_max_msg_size = plni->plni_max_msg_size;
185
186         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: post hello %p", libcfs_id2str(id),
187                        tx->tx_peer->plp_credits,
188                        tx->tx_peer->plp_outstanding_credits,
189                        tx->tx_peer->plp_sent_credits,
190                        plni->plni_peer_credits + 
191                        tx->tx_peer->plp_lazy_credits, tx);
192         ptllnd_post_tx(tx);
193
194         return plp;
195 }
196
197 int
198 ptllnd_count_q(struct list_head *q)
199 {
200         struct list_head *e;
201         int               n = 0;
202         
203         list_for_each(e, q) {
204                 n++;
205         }
206         
207         return n;
208 }
209
210 const char *
211 ptllnd_tx_typestr(int type) 
212 {
213         switch (type) {
214         case PTLLND_RDMA_WRITE:
215                 return "rdma_write";
216                 
217         case PTLLND_RDMA_READ:
218                 return "rdma_read";
219
220         case PTLLND_MSG_TYPE_PUT:
221                 return "put_req";
222                 
223         case PTLLND_MSG_TYPE_GET:
224                 return "get_req";
225
226         case PTLLND_MSG_TYPE_IMMEDIATE:
227                 return "immediate";
228
229         case PTLLND_MSG_TYPE_NOOP:
230                 return "noop";
231
232         case PTLLND_MSG_TYPE_HELLO:
233                 return "hello";
234
235         default:
236                 return "<unknown>";
237         }
238 }
239
240 void
241 ptllnd_debug_tx(ptllnd_tx_t *tx) 
242 {
243         CDEBUG(D_WARNING, "%s %s b %ld.%06ld/%ld.%06ld"
244                " r %ld.%06ld/%ld.%06ld status %d\n",
245                ptllnd_tx_typestr(tx->tx_type),
246                libcfs_id2str(tx->tx_peer->plp_id),
247                tx->tx_bulk_posted.tv_sec, tx->tx_bulk_posted.tv_usec, 
248                tx->tx_bulk_done.tv_sec, tx->tx_bulk_done.tv_usec,
249                tx->tx_req_posted.tv_sec, tx->tx_req_posted.tv_usec,
250                tx->tx_req_done.tv_sec, tx->tx_req_done.tv_usec,
251                tx->tx_status);
252 }
253
254 void
255 ptllnd_debug_peer(lnet_ni_t *ni, lnet_process_id_t id)
256 {
257         ptllnd_peer_t    *plp = ptllnd_find_peer(ni, id, 0);
258         struct list_head *tmp;
259         ptllnd_ni_t      *plni = ni->ni_data;
260         ptllnd_tx_t      *tx;
261         
262         if (plp == NULL) {
263                 CDEBUG(D_WARNING, "No peer %s\n", libcfs_id2str(id));
264                 return;
265         }
266         
267         CDEBUG(D_WARNING, "%s %s%s [%d] "LPU64".%06d m "LPU64" q %d/%d c %d/%d+%d(%d)\n",
268                libcfs_id2str(id), 
269                plp->plp_recvd_hello ? "H" : "_",
270                plp->plp_closing     ? "C" : "_",
271                plp->plp_refcount,
272                plp->plp_stamp / 1000000, (int)(plp->plp_stamp % 1000000),
273                plp->plp_match,
274                ptllnd_count_q(&plp->plp_txq),
275                ptllnd_count_q(&plp->plp_activeq),
276                plp->plp_credits, plp->plp_outstanding_credits, plp->plp_sent_credits,
277                plni->plni_peer_credits + plp->plp_lazy_credits);
278
279         CDEBUG(D_WARNING, "txq:\n");
280         list_for_each (tmp, &plp->plp_txq) {
281                 tx = list_entry(tmp, ptllnd_tx_t, tx_list);
282                 
283                 ptllnd_debug_tx(tx);
284         }
285
286         CDEBUG(D_WARNING, "activeq:\n");
287         list_for_each (tmp, &plp->plp_activeq) {
288                 tx = list_entry(tmp, ptllnd_tx_t, tx_list);
289                 
290                 ptllnd_debug_tx(tx);
291         }
292
293         CDEBUG(D_WARNING, "zombies:\n");
294         list_for_each (tmp, &plni->plni_zombie_txs) {
295                 tx = list_entry(tmp, ptllnd_tx_t, tx_list);
296                 
297                 if (tx->tx_peer->plp_id.nid == id.nid &&
298                     tx->tx_peer->plp_id.pid == id.pid)
299                         ptllnd_debug_tx(tx);
300         }
301         
302         CDEBUG(D_WARNING, "history:\n");
303         list_for_each (tmp, &plni->plni_tx_history) {
304                 tx = list_entry(tmp, ptllnd_tx_t, tx_list);
305                 
306                 if (tx->tx_peer->plp_id.nid == id.nid &&
307                     tx->tx_peer->plp_id.pid == id.pid)
308                         ptllnd_debug_tx(tx);
309         }
310         
311         ptllnd_peer_decref(plp);
312 }
313
314 void
315 ptllnd_dump_debug(lnet_ni_t *ni, lnet_process_id_t id)
316 {
317         ptllnd_debug_peer(ni, id);
318         ptllnd_dump_history();
319 }
320
321 void
322 ptllnd_notify(lnet_ni_t *ni, lnet_nid_t nid, int alive)
323 {
324         lnet_process_id_t  id;
325         ptllnd_peer_t     *peer;
326         time_t             start = cfs_time_current_sec();
327         ptllnd_ni_t       *plni = ni->ni_data;
328         int                w = plni->plni_long_wait;
329
330         /* This is only actually used to connect to routers at startup! */
331         LASSERT(alive);
332
333         id.nid = nid;
334         id.pid = LUSTRE_SRV_LNET_PID;
335         
336         peer = ptllnd_find_peer(ni, id, 1);
337         if (peer == NULL)
338                 return;
339
340         /* wait for the peer to reply */
341         while (!peer->plp_recvd_hello) {
342                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
343                         CWARN("Waited %ds to connect to %s\n",
344                               (int)(cfs_time_current_sec() - start),
345                               libcfs_id2str(id));
346                         w *= 2;
347                 }
348                 
349                 ptllnd_wait(ni, w);
350         }
351         
352         ptllnd_peer_decref(peer);
353 }
354
355 int
356 ptllnd_setasync(lnet_ni_t *ni, lnet_process_id_t id, int nasync)
357 {
358         ptllnd_peer_t *peer = ptllnd_find_peer(ni, id, nasync > 0);
359         int            rc;
360         
361         if (peer == NULL)
362                 return -ENOMEM;
363
364         LASSERT (peer->plp_lazy_credits >= 0);
365         LASSERT (peer->plp_extra_lazy_credits >= 0);
366
367         /* If nasync < 0, we're being told we can reduce the total message
368          * headroom.  We can't do this right now because our peer might already
369          * have credits for the extra buffers, so we just account the extra
370          * headroom in case we need it later and only destroy buffers when the
371          * peer closes.
372          *
373          * Note that the following condition handles this case, where it
374          * actually increases the extra lazy credit counter. */
375
376         if (nasync <= peer->plp_extra_lazy_credits) {
377                 peer->plp_extra_lazy_credits -= nasync;
378                 return 0;
379         }
380
381         LASSERT (nasync > 0);
382
383         nasync -= peer->plp_extra_lazy_credits;
384         peer->plp_extra_lazy_credits = 0;
385         
386         rc = ptllnd_size_buffers(ni, nasync);
387         if (rc == 0) {
388                 peer->plp_lazy_credits += nasync;
389                 peer->plp_outstanding_credits += nasync;
390         }
391
392         return rc;
393 }
394
395 __u32
396 ptllnd_cksum (void *ptr, int nob)
397 {
398         char  *c  = ptr;
399         __u32  sum = 0;
400
401         while (nob-- > 0)
402                 sum = ((sum << 1) | (sum >> 31)) + *c++;
403
404         /* ensure I don't return 0 (== no checksum) */
405         return (sum == 0) ? 1 : sum;
406 }
407
408 ptllnd_tx_t *
409 ptllnd_new_tx(ptllnd_peer_t *peer, int type, int payload_nob)
410 {
411         lnet_ni_t   *ni = peer->plp_ni;
412         ptllnd_ni_t *plni = ni->ni_data;
413         ptllnd_tx_t *tx;
414         int          msgsize;
415
416         CDEBUG(D_NET, "peer=%p type=%d payload=%d\n", peer, type, payload_nob);
417
418         switch (type) {
419         default:
420                 LBUG();
421
422         case PTLLND_RDMA_WRITE:
423         case PTLLND_RDMA_READ:
424                 LASSERT (payload_nob == 0);
425                 msgsize = 0;
426                 break;
427
428         case PTLLND_MSG_TYPE_PUT:
429         case PTLLND_MSG_TYPE_GET:
430                 LASSERT (payload_nob == 0);
431                 msgsize = offsetof(kptl_msg_t, ptlm_u) + 
432                           sizeof(kptl_rdma_msg_t);
433                 break;
434
435         case PTLLND_MSG_TYPE_IMMEDIATE:
436                 msgsize = offsetof(kptl_msg_t,
437                                    ptlm_u.immediate.kptlim_payload[payload_nob]);
438                 break;
439
440         case PTLLND_MSG_TYPE_NOOP:
441                 LASSERT (payload_nob == 0);
442                 msgsize = offsetof(kptl_msg_t, ptlm_u);
443                 break;
444
445         case PTLLND_MSG_TYPE_HELLO:
446                 LASSERT (payload_nob == 0);
447                 msgsize = offsetof(kptl_msg_t, ptlm_u) +
448                           sizeof(kptl_hello_msg_t);
449                 break;
450         }
451
452         msgsize = (msgsize + 7) & ~7;
453         LASSERT (msgsize <= peer->plp_max_msg_size);
454
455         LIBCFS_ALLOC(tx, offsetof(ptllnd_tx_t, tx_msg) + msgsize);
456
457         if (tx == NULL) {
458                 CERROR("Can't allocate msg type %d for %s\n",
459                        type, libcfs_id2str(peer->plp_id));
460                 return NULL;
461         }
462
463         CFS_INIT_LIST_HEAD(&tx->tx_list);
464         tx->tx_peer = peer;
465         tx->tx_type = type;
466         tx->tx_lnetmsg = tx->tx_lnetreplymsg = NULL;
467         tx->tx_niov = 0;
468         tx->tx_iov = NULL;
469         tx->tx_reqmdh = PTL_INVALID_HANDLE;
470         tx->tx_bulkmdh = PTL_INVALID_HANDLE;
471         tx->tx_msgsize = msgsize;
472         tx->tx_completing = 0;
473         tx->tx_status = 0;
474
475         memset(&tx->tx_bulk_posted, 0, sizeof(tx->tx_bulk_posted));
476         memset(&tx->tx_bulk_done, 0, sizeof(tx->tx_bulk_done));
477         memset(&tx->tx_req_posted, 0, sizeof(tx->tx_req_posted));
478         memset(&tx->tx_req_done, 0, sizeof(tx->tx_req_done));
479
480         if (msgsize != 0) {
481                 tx->tx_msg.ptlm_magic = PTLLND_MSG_MAGIC;
482                 tx->tx_msg.ptlm_version = PTLLND_MSG_VERSION;
483                 tx->tx_msg.ptlm_type = type;
484                 tx->tx_msg.ptlm_credits = 0;
485                 tx->tx_msg.ptlm_nob = msgsize;
486                 tx->tx_msg.ptlm_cksum = 0;
487                 tx->tx_msg.ptlm_srcnid = ni->ni_nid;
488                 tx->tx_msg.ptlm_srcstamp = plni->plni_stamp;
489                 tx->tx_msg.ptlm_dstnid = peer->plp_id.nid;
490                 tx->tx_msg.ptlm_dststamp = peer->plp_stamp;
491                 tx->tx_msg.ptlm_srcpid = the_lnet.ln_pid;
492                 tx->tx_msg.ptlm_dstpid = peer->plp_id.pid;
493         }
494
495         ptllnd_peer_addref(peer);
496         plni->plni_ntxs++;
497
498         CDEBUG(D_NET, "tx=%p\n",tx);
499
500         return tx;
501 }
502
503 void
504 ptllnd_abort_tx(ptllnd_tx_t *tx, ptl_handle_md_t *mdh)
505 {
506         ptllnd_peer_t   *peer = tx->tx_peer;
507         lnet_ni_t       *ni = peer->plp_ni;
508         int              rc;
509         time_t           start = cfs_time_current_sec();
510         ptllnd_ni_t     *plni = ni->ni_data;
511         int              w = plni->plni_long_wait;
512
513         while (!PtlHandleIsEqual(*mdh, PTL_INVALID_HANDLE)) {
514                 rc = PtlMDUnlink(*mdh);
515 #ifndef LUSTRE_PORTALS_UNLINK_SEMANTICS
516                 if (rc == PTL_OK) /* unlink successful => no unlinked event */
517                         return;
518                 LASSERT (rc == PTL_MD_IN_USE);
519 #endif
520                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
521                         CWARN("Waited %ds to abort tx to %s\n",
522                               (int)(cfs_time_current_sec() - start),
523                               libcfs_id2str(peer->plp_id));
524                         w *= 2;
525                 }
526                 /* Wait for ptllnd_tx_event() to invalidate */
527                 ptllnd_wait(ni, w);
528         }
529 }
530
531 void
532 ptllnd_cull_tx_history(ptllnd_ni_t *plni)
533 {
534         int max = plni->plni_max_tx_history;
535
536         while (plni->plni_ntx_history > max) {
537                 ptllnd_tx_t *tx = list_entry(plni->plni_tx_history.next, 
538                                              ptllnd_tx_t, tx_list);
539                 list_del(&tx->tx_list);
540
541                 ptllnd_peer_decref(tx->tx_peer);
542
543                 LIBCFS_FREE(tx, offsetof(ptllnd_tx_t, tx_msg) + tx->tx_msgsize);
544
545                 LASSERT (plni->plni_ntxs > 0);
546                 plni->plni_ntxs--;
547                 plni->plni_ntx_history--;
548         }
549 }
550
551 void
552 ptllnd_tx_done(ptllnd_tx_t *tx)
553 {
554         ptllnd_peer_t   *peer = tx->tx_peer;
555         lnet_ni_t       *ni = peer->plp_ni;
556         ptllnd_ni_t     *plni = ni->ni_data;
557
558         /* CAVEAT EMPTOR: If this tx is being aborted, I'll continue to get
559          * events for this tx until it's unlinked.  So I set tx_completing to
560          * flag the tx is getting handled */
561
562         if (tx->tx_completing)
563                 return;
564
565         tx->tx_completing = 1;
566
567         if (!list_empty(&tx->tx_list))
568                 list_del_init(&tx->tx_list);
569
570         if (tx->tx_status != 0) {
571                 if (plni->plni_debug) {
572                         CERROR("Completing tx for %s with error %d\n",
573                                libcfs_id2str(peer->plp_id), tx->tx_status);
574                         ptllnd_debug_tx(tx);
575                 }
576                 ptllnd_close_peer(peer, tx->tx_status);
577         }
578         
579         ptllnd_abort_tx(tx, &tx->tx_reqmdh);
580         ptllnd_abort_tx(tx, &tx->tx_bulkmdh);
581
582         if (tx->tx_niov > 0) {
583                 LIBCFS_FREE(tx->tx_iov, tx->tx_niov * sizeof(*tx->tx_iov));
584                 tx->tx_niov = 0;
585         }
586
587         if (tx->tx_lnetreplymsg != NULL) {
588                 LASSERT (tx->tx_type == PTLLND_MSG_TYPE_GET);
589                 LASSERT (tx->tx_lnetmsg != NULL);
590                 /* Simulate GET success always  */
591                 lnet_finalize(ni, tx->tx_lnetmsg, 0);
592                 CDEBUG(D_NET, "lnet_finalize(tx_lnetreplymsg=%p)\n",tx->tx_lnetreplymsg);
593                 lnet_finalize(ni, tx->tx_lnetreplymsg, tx->tx_status);
594         } else if (tx->tx_lnetmsg != NULL) {
595                 lnet_finalize(ni, tx->tx_lnetmsg, tx->tx_status);
596         }
597
598         plni->plni_ntx_history++;
599         list_add_tail(&tx->tx_list, &plni->plni_tx_history);
600         
601         ptllnd_cull_tx_history(plni);
602 }
603
604 int
605 ptllnd_set_txiov(ptllnd_tx_t *tx,
606                  unsigned int niov, struct iovec *iov,
607                  unsigned int offset, unsigned int len)
608 {
609         ptl_md_iovec_t *piov;
610         int             npiov;
611
612         if (len == 0) {
613                 tx->tx_niov = 0;
614                 return 0;
615         }
616
617         /*
618          * Remove iovec's at the beginning that
619          * are skipped because of the offset.
620          * Adjust the offset accordingly
621          */
622         for (;;) {
623                 LASSERT (niov > 0);
624                 if (offset < iov->iov_len)
625                         break;
626                 offset -= iov->iov_len;
627                 niov--;
628                 iov++;
629         }
630
631         for (;;) {
632                 int temp_offset = offset;
633                 int resid = len;
634                 LIBCFS_ALLOC(piov, niov * sizeof(*piov));
635                 if (piov == NULL)
636                         return -ENOMEM;
637
638                 for (npiov = 0;; npiov++) {
639                         LASSERT (npiov < niov);
640                         LASSERT (iov->iov_len >= temp_offset);
641
642                         piov[npiov].iov_base = iov[npiov].iov_base + temp_offset;
643                         piov[npiov].iov_len = iov[npiov].iov_len - temp_offset;
644                         
645                         if (piov[npiov].iov_len >= resid) {
646                                 piov[npiov].iov_len = resid;
647                                 npiov++;
648                                 break;
649                         }
650                         resid -= piov[npiov].iov_len;
651                         temp_offset = 0;
652                 }
653
654                 if (npiov == niov) {
655                         tx->tx_niov = niov;
656                         tx->tx_iov = piov;
657                         return 0;
658                 }
659
660                 /* Dang! The piov I allocated was too big and it's a drag to
661                  * have to maintain separate 'allocated' and 'used' sizes, so
662                  * I'll just do it again; NB this doesn't happen normally... */
663                 LIBCFS_FREE(piov, niov * sizeof(*piov));
664                 niov = npiov;
665         }
666 }
667
668 void
669 ptllnd_set_md_buffer(ptl_md_t *md, ptllnd_tx_t *tx)
670 {
671         unsigned int    niov = tx->tx_niov;
672         ptl_md_iovec_t *iov = tx->tx_iov;
673
674         LASSERT ((md->options & PTL_MD_IOVEC) == 0);
675
676         if (niov == 0) {
677                 md->start = NULL;
678                 md->length = 0;
679         } else if (niov == 1) {
680                 md->start = iov[0].iov_base;
681                 md->length = iov[0].iov_len;
682         } else {
683                 md->start = iov;
684                 md->length = niov;
685                 md->options |= PTL_MD_IOVEC;
686         }
687 }
688
689 int
690 ptllnd_post_buffer(ptllnd_buffer_t *buf)
691 {
692         lnet_ni_t        *ni = buf->plb_ni;
693         ptllnd_ni_t      *plni = ni->ni_data;
694         ptl_process_id_t  anyid = {
695                 .nid       = PTL_NID_ANY,
696                 .pid       = PTL_PID_ANY};
697         ptl_md_t          md = {
698                 .start     = buf->plb_buffer,
699                 .length    = plni->plni_buffer_size,
700                 .threshold = PTL_MD_THRESH_INF,
701                 .max_size  = plni->plni_max_msg_size,
702                 .options   = (PTLLND_MD_OPTIONS |
703                               PTL_MD_OP_PUT | PTL_MD_MAX_SIZE | 
704                               PTL_MD_LOCAL_ALIGN8),
705                 .user_ptr  = ptllnd_obj2eventarg(buf, PTLLND_EVENTARG_TYPE_BUF),
706                 .eq_handle = plni->plni_eqh};
707         ptl_handle_me_t meh;
708         int             rc;
709
710         LASSERT (!buf->plb_posted);
711
712         rc = PtlMEAttach(plni->plni_nih, plni->plni_portal,
713                          anyid, LNET_MSG_MATCHBITS, 0,
714                          PTL_UNLINK, PTL_INS_AFTER, &meh);
715         if (rc != PTL_OK) {
716                 CERROR("PtlMEAttach failed: %s(%d)\n",
717                        ptllnd_errtype2str(rc), rc);
718                 return -ENOMEM;
719         }
720
721         buf->plb_posted = 1;
722         plni->plni_nposted_buffers++;
723
724         rc = PtlMDAttach(meh, md, LNET_UNLINK, &buf->plb_md);
725         if (rc == PTL_OK)
726                 return 0;
727
728         CERROR("PtlMDAttach failed: %s(%d)\n",
729                ptllnd_errtype2str(rc), rc);
730
731         buf->plb_posted = 0;
732         plni->plni_nposted_buffers--;
733
734         rc = PtlMEUnlink(meh);
735         LASSERT (rc == PTL_OK);
736
737         return -ENOMEM;
738 }
739
740 void
741 ptllnd_check_sends(ptllnd_peer_t *peer)
742 {
743         lnet_ni_t      *ni = peer->plp_ni;
744         ptllnd_ni_t    *plni = ni->ni_data;
745         ptllnd_tx_t    *tx;
746         ptl_md_t        md;
747         ptl_handle_md_t mdh;
748         int             rc;
749
750         CDEBUG(D_NET, "%s: [%d/%d+%d(%d)\n",
751                libcfs_id2str(peer->plp_id), peer->plp_credits,
752                peer->plp_outstanding_credits, peer->plp_sent_credits,
753                plni->plni_peer_credits + peer->plp_lazy_credits);
754
755         if (list_empty(&peer->plp_txq) &&
756             peer->plp_outstanding_credits >= PTLLND_CREDIT_HIGHWATER(plni) &&
757             peer->plp_credits != 0) {
758
759                 tx = ptllnd_new_tx(peer, PTLLND_MSG_TYPE_NOOP, 0);
760                 CDEBUG(D_NET, "NOOP tx=%p\n",tx);
761                 if (tx == NULL) {
762                         CERROR("Can't return credits to %s\n",
763                                libcfs_id2str(peer->plp_id));
764                 } else {
765                         list_add_tail(&tx->tx_list, &peer->plp_txq);
766                 }
767         }
768
769         while (!list_empty(&peer->plp_txq)) {
770                 tx = list_entry(peer->plp_txq.next, ptllnd_tx_t, tx_list);
771
772                 LASSERT (tx->tx_msgsize > 0);
773
774                 LASSERT (peer->plp_outstanding_credits >= 0);
775                 LASSERT (peer->plp_sent_credits >= 0);
776                 LASSERT (peer->plp_outstanding_credits + peer->plp_sent_credits
777                          <= plni->plni_peer_credits + peer->plp_lazy_credits);
778                 LASSERT (peer->plp_credits >= 0);
779
780                 if (peer->plp_credits == 0) {   /* no credits */
781                         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: no creds for %p",
782                                        libcfs_id2str(peer->plp_id),
783                                        peer->plp_credits,
784                                        peer->plp_outstanding_credits,
785                                        peer->plp_sent_credits,
786                                        plni->plni_peer_credits +
787                                        peer->plp_lazy_credits, tx);
788                         break;
789                 }
790                 
791                 if (peer->plp_credits == 1 &&   /* last credit reserved for */
792                     peer->plp_outstanding_credits == 0) { /* returning credits */
793                         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: too few creds for %p",
794                                        libcfs_id2str(peer->plp_id),
795                                        peer->plp_credits,
796                                        peer->plp_outstanding_credits,
797                                        peer->plp_sent_credits,
798                                        plni->plni_peer_credits +
799                                        peer->plp_lazy_credits, tx);
800                         break;
801                 }
802                 
803                 list_del(&tx->tx_list);
804                 list_add_tail(&tx->tx_list, &peer->plp_activeq);
805
806                 CDEBUG(D_NET, "Sending at TX=%p type=%s (%d)\n",tx,
807                         ptllnd_msgtype2str(tx->tx_type),tx->tx_type);
808
809                 if (tx->tx_type == PTLLND_MSG_TYPE_NOOP &&
810                     (!list_empty(&peer->plp_txq) ||
811                      peer->plp_outstanding_credits <
812                      PTLLND_CREDIT_HIGHWATER(plni))) {
813                         /* redundant NOOP */
814                         ptllnd_tx_done(tx);
815                         continue;
816                 }
817
818                 /* Set stamp at the last minute; on a new peer, I don't know it
819                  * until I receive the HELLO back */
820                 tx->tx_msg.ptlm_dststamp = peer->plp_stamp;
821
822                 /*
823                  * Return all the credits we have
824                  */
825                 tx->tx_msg.ptlm_credits = peer->plp_outstanding_credits;
826                 peer->plp_sent_credits += peer->plp_outstanding_credits;
827                 peer->plp_outstanding_credits = 0;
828
829                 /*
830                  * One less credit
831                  */
832                 peer->plp_credits--;
833
834                 if (plni->plni_checksum)
835                         tx->tx_msg.ptlm_cksum = 
836                                 ptllnd_cksum(&tx->tx_msg,
837                                              offsetof(kptl_msg_t, ptlm_u));
838
839                 md.user_ptr = ptllnd_obj2eventarg(tx, PTLLND_EVENTARG_TYPE_TX);
840                 md.eq_handle = plni->plni_eqh;
841                 md.threshold = 1;
842                 md.options = PTLLND_MD_OPTIONS;
843                 md.start = &tx->tx_msg;
844                 md.length = tx->tx_msgsize;
845
846                 rc = PtlMDBind(plni->plni_nih, md, LNET_UNLINK, &mdh);
847                 if (rc != PTL_OK) {
848                         CERROR("PtlMDBind for %s failed: %s(%d)\n",
849                                libcfs_id2str(peer->plp_id),
850                                ptllnd_errtype2str(rc), rc);
851                         tx->tx_status = -EIO;
852                         ptllnd_tx_done(tx);
853                         break;
854                 }
855
856                 LASSERT (tx->tx_type != PTLLND_RDMA_WRITE &&
857                          tx->tx_type != PTLLND_RDMA_READ);
858                 
859                 tx->tx_reqmdh = mdh;
860                 gettimeofday(&tx->tx_req_posted, NULL);
861
862                 PTLLND_HISTORY("%s[%d/%d+%d(%d)]: %s %p c %d",
863                                libcfs_id2str(peer->plp_id),
864                                peer->plp_credits,
865                                peer->plp_outstanding_credits,
866                                peer->plp_sent_credits,
867                                plni->plni_peer_credits +
868                                peer->plp_lazy_credits,
869                                ptllnd_msgtype2str(tx->tx_type), tx,
870                                tx->tx_msg.ptlm_credits);
871
872                 rc = PtlPut(mdh, PTL_NOACK_REQ, peer->plp_ptlid,
873                             plni->plni_portal, 0, LNET_MSG_MATCHBITS, 0, 0);
874                 if (rc != PTL_OK) {
875                         CERROR("PtlPut for %s failed: %s(%d)\n",
876                                libcfs_id2str(peer->plp_id),
877                                ptllnd_errtype2str(rc), rc);
878                         tx->tx_status = -EIO;
879                         ptllnd_tx_done(tx);
880                         break;
881                 }
882         }
883 }
884
885 int
886 ptllnd_passive_rdma(ptllnd_peer_t *peer, int type, lnet_msg_t *msg,
887                     unsigned int niov, struct iovec *iov,
888                     unsigned int offset, unsigned int len)
889 {
890         lnet_ni_t      *ni = peer->plp_ni;
891         ptllnd_ni_t    *plni = ni->ni_data;
892         ptllnd_tx_t    *tx = ptllnd_new_tx(peer, type, 0);
893         __u64           matchbits;
894         ptl_md_t        md;
895         ptl_handle_md_t mdh;
896         ptl_handle_me_t meh;
897         int             rc;
898         int             rc2;
899         time_t          start;
900         int             w;
901
902         CDEBUG(D_NET, "niov=%d offset=%d len=%d\n",niov,offset,len);
903
904         LASSERT (type == PTLLND_MSG_TYPE_GET ||
905                  type == PTLLND_MSG_TYPE_PUT);
906
907         if (tx == NULL) {
908                 CERROR("Can't allocate %s tx for %s\n",
909                        type == PTLLND_MSG_TYPE_GET ? "GET" : "PUT/REPLY",
910                        libcfs_id2str(peer->plp_id));
911                 return -ENOMEM;
912         }
913
914         rc = ptllnd_set_txiov(tx, niov, iov, offset, len);
915         if (rc != 0) {
916                 CERROR ("Can't allocate iov %d for %s\n",
917                         niov, libcfs_id2str(peer->plp_id));
918                 rc = -ENOMEM;
919                 goto failed;
920         }
921
922         md.user_ptr = ptllnd_obj2eventarg(tx, PTLLND_EVENTARG_TYPE_TX);
923         md.eq_handle = plni->plni_eqh;
924         md.threshold = 1;
925         md.max_size = 0;
926         md.options = PTLLND_MD_OPTIONS;
927         if(type == PTLLND_MSG_TYPE_GET)
928                 md.options |= PTL_MD_OP_PUT | PTL_MD_ACK_DISABLE;
929         else
930                 md.options |= PTL_MD_OP_GET;
931         ptllnd_set_md_buffer(&md, tx);
932
933         start = cfs_time_current_sec();
934         w = plni->plni_long_wait;
935
936         while (!peer->plp_recvd_hello) {        /* wait to validate plp_match */
937                 if (peer->plp_closing) {
938                         rc = -EIO;
939                         goto failed;
940                 }
941                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
942                         CWARN("Waited %ds to connect to %s\n",
943                               (int)(cfs_time_current_sec() - start),
944                               libcfs_id2str(peer->plp_id));
945                         w *= 2;
946                 }
947                 ptllnd_wait(ni, w);
948         }
949
950         if (peer->plp_match < PTL_RESERVED_MATCHBITS)
951                 peer->plp_match = PTL_RESERVED_MATCHBITS;
952         matchbits = peer->plp_match++;
953
954         rc = PtlMEAttach(plni->plni_nih, plni->plni_portal, peer->plp_ptlid,
955                          matchbits, 0, PTL_UNLINK, PTL_INS_BEFORE, &meh);
956         if (rc != PTL_OK) {
957                 CERROR("PtlMEAttach for %s failed: %s(%d)\n",
958                        libcfs_id2str(peer->plp_id),
959                        ptllnd_errtype2str(rc), rc);
960                 rc = -EIO;
961                 goto failed;
962         }
963
964         gettimeofday(&tx->tx_bulk_posted, NULL);
965
966         rc = PtlMDAttach(meh, md, LNET_UNLINK, &mdh);
967         if (rc != PTL_OK) {
968                 CERROR("PtlMDAttach for %s failed: %s(%d)\n",
969                        libcfs_id2str(peer->plp_id),
970                        ptllnd_errtype2str(rc), rc);
971                 rc2 = PtlMEUnlink(meh);
972                 LASSERT (rc2 == PTL_OK);
973                 rc = -EIO;
974                 goto failed;
975         }
976         tx->tx_bulkmdh = mdh;
977
978         /*
979          * We need to set the stamp here because it
980          * we could have received a HELLO above that set
981          * peer->plp_stamp
982          */
983         tx->tx_msg.ptlm_dststamp = peer->plp_stamp;
984
985         tx->tx_msg.ptlm_u.rdma.kptlrm_hdr = msg->msg_hdr;
986         tx->tx_msg.ptlm_u.rdma.kptlrm_matchbits = matchbits;
987
988         if (type == PTLLND_MSG_TYPE_GET) {
989                 tx->tx_lnetreplymsg = lnet_create_reply_msg(ni, msg);
990                 if (tx->tx_lnetreplymsg == NULL) {
991                         CERROR("Can't create reply for GET to %s\n",
992                                libcfs_id2str(msg->msg_target));
993                         rc = -ENOMEM;
994                         goto failed;
995                 }
996         }
997
998         tx->tx_lnetmsg = msg;
999         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: post passive %s p %d %p",
1000                        libcfs_id2str(msg->msg_target),
1001                        peer->plp_credits, peer->plp_outstanding_credits,
1002                        peer->plp_sent_credits,
1003                        plni->plni_peer_credits + peer->plp_lazy_credits,
1004                        lnet_msgtyp2str(msg->msg_type),
1005                        (le32_to_cpu(msg->msg_type) == LNET_MSG_PUT) ? 
1006                        le32_to_cpu(msg->msg_hdr.msg.put.ptl_index) :
1007                        (le32_to_cpu(msg->msg_type) == LNET_MSG_GET) ? 
1008                        le32_to_cpu(msg->msg_hdr.msg.get.ptl_index) : -1,
1009                        tx);
1010         ptllnd_post_tx(tx);
1011         return 0;
1012
1013  failed:
1014         ptllnd_tx_done(tx);
1015         return rc;
1016 }
1017
1018 int
1019 ptllnd_active_rdma(ptllnd_peer_t *peer, int type,
1020                    lnet_msg_t *msg, __u64 matchbits,
1021                    unsigned int niov, struct iovec *iov,
1022                    unsigned int offset, unsigned int len)
1023 {
1024         lnet_ni_t       *ni = peer->plp_ni;
1025         ptllnd_ni_t     *plni = ni->ni_data;
1026         ptllnd_tx_t     *tx = ptllnd_new_tx(peer, type, 0);
1027         ptl_md_t         md;
1028         ptl_handle_md_t  mdh;
1029         int              rc;
1030
1031         LASSERT (type == PTLLND_RDMA_READ ||
1032                  type == PTLLND_RDMA_WRITE);
1033
1034         if (tx == NULL) {
1035                 CERROR("Can't allocate tx for RDMA %s with %s\n",
1036                        (type == PTLLND_RDMA_WRITE) ? "write" : "read",
1037                        libcfs_id2str(peer->plp_id));
1038                 ptllnd_close_peer(peer, -ENOMEM);
1039                 return -ENOMEM;
1040         }
1041
1042         rc = ptllnd_set_txiov(tx, niov, iov, offset, len);
1043         if (rc != 0) {
1044                 CERROR ("Can't allocate iov %d for %s\n",
1045                         niov, libcfs_id2str(peer->plp_id));
1046                 rc = -ENOMEM;
1047                 goto failed;
1048         }
1049
1050         md.user_ptr = ptllnd_obj2eventarg(tx, PTLLND_EVENTARG_TYPE_TX);
1051         md.eq_handle = plni->plni_eqh;
1052         md.max_size = 0;
1053         md.options = PTLLND_MD_OPTIONS;
1054         md.threshold = (type == PTLLND_RDMA_READ) ? 2 : 1;
1055
1056         ptllnd_set_md_buffer(&md, tx);
1057
1058         rc = PtlMDBind(plni->plni_nih, md, LNET_UNLINK, &mdh);
1059         if (rc != PTL_OK) {
1060                 CERROR("PtlMDBind for %s failed: %s(%d)\n",
1061                        libcfs_id2str(peer->plp_id),
1062                        ptllnd_errtype2str(rc), rc);
1063                 rc = -EIO;
1064                 goto failed;
1065         }
1066
1067         tx->tx_bulkmdh = mdh;
1068         tx->tx_lnetmsg = msg;
1069
1070         ptllnd_set_tx_deadline(tx);
1071         list_add_tail(&tx->tx_list, &peer->plp_activeq);
1072         gettimeofday(&tx->tx_bulk_posted, NULL);
1073
1074         if (type == PTLLND_RDMA_READ)
1075                 rc = PtlGet(mdh, peer->plp_ptlid,
1076                             plni->plni_portal, 0, matchbits, 0);
1077         else
1078                 rc = PtlPut(mdh, PTL_NOACK_REQ, peer->plp_ptlid,
1079                             plni->plni_portal, 0, matchbits, 0, 
1080                             (msg == NULL) ? PTLLND_RDMA_FAIL : PTLLND_RDMA_OK);
1081
1082         if (rc == PTL_OK)
1083                 return 0;
1084
1085         CERROR("Can't initiate RDMA with %s: %s(%d)\n",
1086                libcfs_id2str(peer->plp_id),
1087                ptllnd_errtype2str(rc), rc);
1088
1089         tx->tx_lnetmsg = NULL;
1090  failed:
1091         tx->tx_status = rc;
1092         ptllnd_tx_done(tx);    /* this will close peer */
1093         return rc;
1094 }
1095
1096 int
1097 ptllnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *msg)
1098 {
1099         ptllnd_ni_t    *plni = ni->ni_data;
1100         ptllnd_peer_t  *plp;
1101         ptllnd_tx_t    *tx;
1102         int             nob;
1103         int             rc;
1104
1105         LASSERT (!msg->msg_routing);
1106         LASSERT (msg->msg_kiov == NULL);
1107
1108         LASSERT (msg->msg_niov <= PTL_MD_MAX_IOV); /* !!! */
1109
1110         CDEBUG(D_NET, "%s [%d]+%d,%d -> %s%s\n", 
1111                lnet_msgtyp2str(msg->msg_type),
1112                msg->msg_niov, msg->msg_offset, msg->msg_len,
1113                libcfs_nid2str(msg->msg_target.nid),
1114                msg->msg_target_is_router ? "(rtr)" : "");
1115
1116         if ((msg->msg_target.pid & LNET_PID_USERFLAG) != 0) {
1117                 CERROR("Can't send to non-kernel peer %s\n",
1118                        libcfs_id2str(msg->msg_target));
1119                 return -EHOSTUNREACH;
1120         }
1121         
1122         plp = ptllnd_find_peer(ni, msg->msg_target, 1);
1123         if (plp == NULL)
1124                 return -ENOMEM;
1125
1126         switch (msg->msg_type) {
1127         default:
1128                 LBUG();
1129
1130         case LNET_MSG_ACK:
1131                 LASSERT (msg->msg_len == 0);
1132                 break;                          /* send IMMEDIATE */
1133
1134         case LNET_MSG_GET:
1135                 if (msg->msg_target_is_router)
1136                         break;                  /* send IMMEDIATE */
1137
1138                 nob = msg->msg_md->md_length;
1139                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[nob]);
1140                 if (nob <= plni->plni_max_msg_size)
1141                         break;
1142
1143                 LASSERT ((msg->msg_md->md_options & LNET_MD_KIOV) == 0);
1144                 rc = ptllnd_passive_rdma(plp, PTLLND_MSG_TYPE_GET, msg,
1145                                          msg->msg_md->md_niov,
1146                                          msg->msg_md->md_iov.iov,
1147                                          0, msg->msg_md->md_length);
1148                 ptllnd_peer_decref(plp);
1149                 return rc;
1150
1151         case LNET_MSG_REPLY:
1152         case LNET_MSG_PUT:
1153                 nob = msg->msg_len;
1154                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[nob]);
1155                 if (nob <= plp->plp_max_msg_size)
1156                         break;                  /* send IMMEDIATE */
1157
1158                 rc = ptllnd_passive_rdma(plp, PTLLND_MSG_TYPE_PUT, msg,
1159                                          msg->msg_niov, msg->msg_iov,
1160                                          msg->msg_offset, msg->msg_len);
1161                 ptllnd_peer_decref(plp);
1162                 return rc;
1163         }
1164
1165         /* send IMMEDIATE
1166          * NB copy the payload so we don't have to do a fragmented send */
1167
1168         tx = ptllnd_new_tx(plp, PTLLND_MSG_TYPE_IMMEDIATE, msg->msg_len);
1169         if (tx == NULL) {
1170                 CERROR("Can't allocate tx for lnet type %d to %s\n",
1171                        msg->msg_type, libcfs_id2str(msg->msg_target));
1172                 ptllnd_peer_decref(plp);
1173                 return -ENOMEM;
1174         }
1175
1176         lnet_copy_iov2flat(tx->tx_msgsize, &tx->tx_msg,
1177                            offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload),
1178                            msg->msg_niov, msg->msg_iov, msg->msg_offset,
1179                            msg->msg_len);
1180         tx->tx_msg.ptlm_u.immediate.kptlim_hdr = msg->msg_hdr;
1181
1182         tx->tx_lnetmsg = msg;
1183         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: post immediate %s p %d %p",
1184                        libcfs_id2str(msg->msg_target),
1185                        plp->plp_credits, plp->plp_outstanding_credits,
1186                        plp->plp_sent_credits,
1187                        plni->plni_peer_credits + plp->plp_lazy_credits,
1188                        lnet_msgtyp2str(msg->msg_type),
1189                        (le32_to_cpu(msg->msg_type) == LNET_MSG_PUT) ? 
1190                        le32_to_cpu(msg->msg_hdr.msg.put.ptl_index) :
1191                        (le32_to_cpu(msg->msg_type) == LNET_MSG_GET) ? 
1192                        le32_to_cpu(msg->msg_hdr.msg.get.ptl_index) : -1,
1193                        tx);
1194         ptllnd_post_tx(tx);
1195         ptllnd_peer_decref(plp);
1196         return 0;
1197 }
1198
1199 void
1200 ptllnd_rx_done(ptllnd_rx_t *rx)
1201 {
1202         ptllnd_peer_t *plp = rx->rx_peer;
1203         lnet_ni_t     *ni = plp->plp_ni;
1204         ptllnd_ni_t   *plni = ni->ni_data;
1205
1206         plp->plp_outstanding_credits++;
1207
1208         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: rx=%p done\n",
1209                        libcfs_id2str(plp->plp_id),
1210                        plp->plp_credits, plp->plp_outstanding_credits, 
1211                        plp->plp_sent_credits,
1212                        plni->plni_peer_credits + plp->plp_lazy_credits, rx);
1213
1214         ptllnd_check_sends(rx->rx_peer);
1215
1216         LASSERT (plni->plni_nrxs > 0);
1217         plni->plni_nrxs--;
1218 }
1219
1220 int
1221 ptllnd_eager_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg,
1222                   void **new_privatep)
1223 {
1224         /* Shouldn't get here; recvs only block for router buffers */
1225         LBUG();
1226         return 0;
1227 }
1228
1229 int
1230 ptllnd_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg,
1231             int delayed, unsigned int niov,
1232             struct iovec *iov, lnet_kiov_t *kiov,
1233             unsigned int offset, unsigned int mlen, unsigned int rlen)
1234 {
1235         ptllnd_rx_t    *rx = private;
1236         int             rc = 0;
1237         int             nob;
1238
1239         LASSERT (kiov == NULL);
1240         LASSERT (niov <= PTL_MD_MAX_IOV);       /* !!! */
1241
1242         switch (rx->rx_msg->ptlm_type) {
1243         default:
1244                 LBUG();
1245
1246         case PTLLND_MSG_TYPE_IMMEDIATE:
1247                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[mlen]);
1248                 if (nob > rx->rx_nob) {
1249                         CERROR("Immediate message from %s too big: %d(%d)\n",
1250                                libcfs_id2str(rx->rx_peer->plp_id),
1251                                nob, rx->rx_nob);
1252                         rc = -EPROTO;
1253                         break;
1254                 }
1255                 lnet_copy_flat2iov(niov, iov, offset,
1256                                    rx->rx_nob, rx->rx_msg,
1257                                    offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload),
1258                                    mlen);
1259                 lnet_finalize(ni, msg, 0);
1260                 break;
1261
1262         case PTLLND_MSG_TYPE_PUT:
1263                 rc = ptllnd_active_rdma(rx->rx_peer, PTLLND_RDMA_READ, msg,
1264                                         rx->rx_msg->ptlm_u.rdma.kptlrm_matchbits,
1265                                         niov, iov, offset, mlen);
1266                 break;
1267
1268         case PTLLND_MSG_TYPE_GET:
1269                 if (msg != NULL)
1270                         rc = ptllnd_active_rdma(rx->rx_peer, PTLLND_RDMA_WRITE, msg,
1271                                                 rx->rx_msg->ptlm_u.rdma.kptlrm_matchbits,
1272                                                 msg->msg_niov, msg->msg_iov,
1273                                                 msg->msg_offset, msg->msg_len);
1274                 else
1275                         rc = ptllnd_active_rdma(rx->rx_peer, PTLLND_RDMA_WRITE, NULL,
1276                                                 rx->rx_msg->ptlm_u.rdma.kptlrm_matchbits,
1277                                                 0, NULL, 0, 0);
1278                 break;
1279         }
1280
1281         ptllnd_rx_done(rx);
1282         return rc;
1283 }
1284
1285 void
1286 ptllnd_parse_request(lnet_ni_t *ni, ptl_process_id_t initiator,
1287                      kptl_msg_t *msg, unsigned int nob)
1288 {
1289         ptllnd_ni_t      *plni = ni->ni_data;
1290         const int         basenob = offsetof(kptl_msg_t, ptlm_u);
1291         lnet_process_id_t srcid;
1292         ptllnd_rx_t       rx;
1293         int               flip;
1294         __u16             msg_version;
1295         __u32             msg_cksum;
1296         ptllnd_peer_t    *plp;
1297         int               rc;
1298
1299         if (nob < 6) {
1300                 CERROR("Very short receive from %s\n",
1301                        ptllnd_ptlid2str(initiator));
1302                 return;
1303         }
1304
1305         /* I can at least read MAGIC/VERSION */
1306
1307         flip = msg->ptlm_magic == __swab32(PTLLND_MSG_MAGIC);
1308         if (!flip && msg->ptlm_magic != PTLLND_MSG_MAGIC) {
1309                 CERROR("Bad protocol magic %08x from %s\n", 
1310                        msg->ptlm_magic, ptllnd_ptlid2str(initiator));
1311                 return;
1312         }
1313
1314         msg_version = flip ? __swab16(msg->ptlm_version) : msg->ptlm_version;
1315
1316         if (msg_version != PTLLND_MSG_VERSION) {
1317                 CERROR("Bad protocol version %04x from %s: %04x expected\n", 
1318                        (__u32)msg_version, ptllnd_ptlid2str(initiator), PTLLND_MSG_VERSION);
1319
1320                 if (plni->plni_abort_on_protocol_mismatch)
1321                         abort();
1322
1323                 return;
1324         }
1325
1326         if (nob < basenob) {
1327                 CERROR("Short receive from %s: got %d, wanted at least %d\n",
1328                        ptllnd_ptlid2str(initiator), nob, basenob);
1329                 return;
1330         }
1331
1332         /* checksum must be computed with
1333          * 1) ptlm_cksum zero and
1334          * 2) BEFORE anything gets modified/flipped
1335          */
1336         msg_cksum = flip ? __swab32(msg->ptlm_cksum) : msg->ptlm_cksum;
1337         msg->ptlm_cksum = 0;
1338         if (msg_cksum != 0 &&
1339             msg_cksum != ptllnd_cksum(msg, offsetof(kptl_msg_t, ptlm_u))) {
1340                 CERROR("Bad checksum from %s\n", ptllnd_ptlid2str(initiator));
1341                 return;
1342         }
1343
1344         msg->ptlm_version = msg_version;
1345         msg->ptlm_cksum = msg_cksum;
1346         
1347         if (flip) {
1348                 /* NB stamps are opaque cookies */
1349                 __swab32s(&msg->ptlm_nob);
1350                 __swab64s(&msg->ptlm_srcnid);
1351                 __swab64s(&msg->ptlm_dstnid);
1352                 __swab32s(&msg->ptlm_srcpid);
1353                 __swab32s(&msg->ptlm_dstpid);
1354         }
1355         
1356         srcid.nid = msg->ptlm_srcnid;
1357         srcid.pid = msg->ptlm_srcpid;
1358
1359         if (LNET_NIDNET(msg->ptlm_srcnid) != LNET_NIDNET(ni->ni_nid)) {
1360                 CERROR("Bad source id %s from %s\n",
1361                        libcfs_id2str(srcid),
1362                        ptllnd_ptlid2str(initiator));
1363                 return;
1364         }
1365
1366         if (msg->ptlm_type == PTLLND_MSG_TYPE_NAK) {
1367                 CERROR("NAK from %s (%s)\n", 
1368                        libcfs_id2str(srcid),
1369                        ptllnd_ptlid2str(initiator));
1370
1371                 if (plni->plni_dump_on_nak)
1372                         ptllnd_dump_debug(ni, srcid);
1373                 
1374                 if (plni->plni_abort_on_nak)
1375                         abort();
1376                 
1377                 return;
1378         }
1379         
1380         if (msg->ptlm_dstnid != ni->ni_nid ||
1381             msg->ptlm_dstpid != the_lnet.ln_pid) {
1382                 CERROR("Bad dstid %s (%s expected) from %s\n",
1383                        libcfs_id2str((lnet_process_id_t) {
1384                                .nid = msg->ptlm_dstnid,
1385                                .pid = msg->ptlm_dstpid}),
1386                        libcfs_id2str((lnet_process_id_t) {
1387                                .nid = ni->ni_nid,
1388                                .pid = the_lnet.ln_pid}),
1389                        libcfs_id2str(srcid));
1390                 return;
1391         }
1392
1393         if (msg->ptlm_dststamp != plni->plni_stamp) {
1394                 CERROR("Bad dststamp "LPX64"("LPX64" expected) from %s\n",
1395                        msg->ptlm_dststamp, plni->plni_stamp,
1396                        libcfs_id2str(srcid));
1397                 return;
1398         }
1399
1400         PTLLND_HISTORY("RX %s: %s %d %p", libcfs_id2str(srcid), 
1401                        ptllnd_msgtype2str(msg->ptlm_type),
1402                        msg->ptlm_credits, &rx);
1403
1404         switch (msg->ptlm_type) {
1405         case PTLLND_MSG_TYPE_PUT:
1406         case PTLLND_MSG_TYPE_GET:
1407                 if (nob < basenob + sizeof(kptl_rdma_msg_t)) {
1408                         CERROR("Short rdma request from %s(%s)\n",
1409                                libcfs_id2str(srcid),
1410                                ptllnd_ptlid2str(initiator));
1411                         return;
1412                 }
1413                 if (flip)
1414                         __swab64s(&msg->ptlm_u.rdma.kptlrm_matchbits);
1415                 break;
1416
1417         case PTLLND_MSG_TYPE_IMMEDIATE:
1418                 if (nob < offsetof(kptl_msg_t,
1419                                    ptlm_u.immediate.kptlim_payload)) {
1420                         CERROR("Short immediate from %s(%s)\n",
1421                                libcfs_id2str(srcid),
1422                                ptllnd_ptlid2str(initiator));
1423                         return;
1424                 }
1425                 break;
1426
1427         case PTLLND_MSG_TYPE_HELLO:
1428                 if (nob < basenob + sizeof(kptl_hello_msg_t)) {
1429                         CERROR("Short hello from %s(%s)\n",
1430                                libcfs_id2str(srcid),
1431                                ptllnd_ptlid2str(initiator));
1432                         return;
1433                 }
1434                 if(flip){
1435                         __swab64s(&msg->ptlm_u.hello.kptlhm_matchbits);
1436                         __swab32s(&msg->ptlm_u.hello.kptlhm_max_msg_size);
1437                 }
1438                 break;
1439                 
1440         case PTLLND_MSG_TYPE_NOOP:
1441                 break;
1442
1443         default:
1444                 CERROR("Bad message type %d from %s(%s)\n", msg->ptlm_type,
1445                        libcfs_id2str(srcid),
1446                        ptllnd_ptlid2str(initiator));
1447                 return;
1448         }
1449
1450         plp = ptllnd_find_peer(ni, srcid, 0);
1451         if (plp == NULL) {
1452                 CERROR("Can't find peer %s\n", libcfs_id2str(srcid));
1453                 return;
1454         }
1455
1456         if (msg->ptlm_type == PTLLND_MSG_TYPE_HELLO) {
1457                 if (plp->plp_recvd_hello) {
1458                         CERROR("Unexpected HELLO from %s\n",
1459                                libcfs_id2str(srcid));
1460                         ptllnd_peer_decref(plp);
1461                         return;
1462                 }
1463
1464                 plp->plp_max_msg_size = msg->ptlm_u.hello.kptlhm_max_msg_size;
1465                 plp->plp_match = msg->ptlm_u.hello.kptlhm_matchbits;
1466                 plp->plp_stamp = msg->ptlm_srcstamp;
1467                 plp->plp_recvd_hello = 1;
1468
1469         } else if (!plp->plp_recvd_hello) {
1470
1471                 CERROR("Bad message type %d (HELLO expected) from %s\n",
1472                        msg->ptlm_type, libcfs_id2str(srcid));
1473                 ptllnd_peer_decref(plp);
1474                 return;
1475
1476         } else if (msg->ptlm_srcstamp != plp->plp_stamp) {
1477
1478                 CERROR("Bad srcstamp "LPX64"("LPX64" expected) from %s\n",
1479                        msg->ptlm_srcstamp, plp->plp_stamp,
1480                        libcfs_id2str(srcid));
1481                 ptllnd_peer_decref(plp);
1482                 return;
1483         }
1484
1485         /* Check peer only sends when I've sent her credits */
1486         if (plp->plp_sent_credits == 0) {
1487                 CERROR("%s[%d/%d+%d(%d)]: unexpected message\n",
1488                        libcfs_id2str(plp->plp_id),
1489                        plp->plp_credits, plp->plp_outstanding_credits, 
1490                        plp->plp_sent_credits,
1491                        plni->plni_peer_credits + plp->plp_lazy_credits);
1492                 return;
1493         }
1494         plp->plp_sent_credits--;
1495         
1496         /* No check for credit overflow - the peer may post new buffers after
1497          * the startup handshake. */
1498         if (msg->ptlm_credits > 0) {
1499                 plp->plp_credits += msg->ptlm_credits;
1500                 ptllnd_check_sends(plp);
1501         }
1502
1503         /* All OK so far; assume the message is good... */
1504
1505         rx.rx_peer      = plp;
1506         rx.rx_msg       = msg;
1507         rx.rx_nob       = nob;
1508         plni->plni_nrxs++;
1509
1510         switch (msg->ptlm_type) {
1511         default: /* message types have been checked already */
1512                 ptllnd_rx_done(&rx);
1513                 break;
1514
1515         case PTLLND_MSG_TYPE_PUT:
1516         case PTLLND_MSG_TYPE_GET:
1517                 rc = lnet_parse(ni, &msg->ptlm_u.rdma.kptlrm_hdr,
1518                                 msg->ptlm_srcnid, &rx, 1);
1519                 if (rc < 0)
1520                         ptllnd_rx_done(&rx);
1521                 break;
1522
1523         case PTLLND_MSG_TYPE_IMMEDIATE:
1524                 rc = lnet_parse(ni, &msg->ptlm_u.immediate.kptlim_hdr,
1525                                 msg->ptlm_srcnid, &rx, 0);
1526                 if (rc < 0)
1527                         ptllnd_rx_done(&rx);
1528                 break;
1529         }
1530
1531         ptllnd_peer_decref(plp);
1532 }
1533
1534 void
1535 ptllnd_buf_event (lnet_ni_t *ni, ptl_event_t *event)
1536 {
1537         ptllnd_buffer_t *buf = ptllnd_eventarg2obj(event->md.user_ptr);
1538         ptllnd_ni_t     *plni = ni->ni_data;
1539         char            *msg = &buf->plb_buffer[event->offset];
1540         int              repost;
1541         int              unlinked = event->type == PTL_EVENT_UNLINK;
1542
1543         LASSERT (buf->plb_ni == ni);
1544         LASSERT (event->type == PTL_EVENT_PUT_END ||
1545                  event->type == PTL_EVENT_UNLINK);
1546
1547         if (event->ni_fail_type != PTL_NI_OK) {
1548
1549                 CERROR("event type %s(%d), status %s(%d) from %s\n",
1550                        ptllnd_evtype2str(event->type), event->type,
1551                        ptllnd_errtype2str(event->ni_fail_type), 
1552                        event->ni_fail_type,
1553                        ptllnd_ptlid2str(event->initiator));
1554
1555         } else if (event->type == PTL_EVENT_PUT_END) {
1556 #if (PTL_MD_LOCAL_ALIGN8 == 0)
1557                 /* Portals can't force message alignment - someone sending an
1558                  * odd-length message could misalign subsequent messages */
1559                 if ((event->mlength & 7) != 0) {
1560                         CERROR("Message from %s has odd length %llu: "
1561                                "probable version incompatibility\n",
1562                                ptllnd_ptlid2str(event->initiator),
1563                                event->mlength);
1564                         LBUG();
1565                 }
1566 #endif
1567                 LASSERT ((event->offset & 7) == 0);
1568
1569                 ptllnd_parse_request(ni, event->initiator,
1570                                      (kptl_msg_t *)msg, event->mlength);
1571         }
1572
1573 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
1574         /* UNLINK event only on explicit unlink */
1575         repost = (event->unlinked && event->type != PTL_EVENT_UNLINK);
1576         if (event->unlinked)
1577                 unlinked = 1;
1578 #else
1579         /* UNLINK event only on implicit unlink */
1580         repost = (event->type == PTL_EVENT_UNLINK);
1581 #endif
1582
1583         if (unlinked) {
1584                 LASSERT(buf->plb_posted);
1585                 buf->plb_posted = 0;
1586                 plni->plni_nposted_buffers--;
1587         }
1588
1589         if (repost)
1590                 (void) ptllnd_post_buffer(buf);
1591 }
1592
1593 void
1594 ptllnd_tx_event (lnet_ni_t *ni, ptl_event_t *event)
1595 {
1596         ptllnd_ni_t *plni = ni->ni_data;
1597         ptllnd_tx_t *tx = ptllnd_eventarg2obj(event->md.user_ptr);
1598         int          error = (event->ni_fail_type != PTL_NI_OK);
1599         int          isreq;
1600         int          isbulk;
1601 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
1602         int          unlinked = event->unlinked;
1603 #else
1604         int          unlinked = (event->type == PTL_EVENT_UNLINK);
1605 #endif
1606
1607         if (error)
1608                 CERROR("Error %s(%d) event %s(%d) unlinked %d, %s(%d) for %s\n",
1609                        ptllnd_errtype2str(event->ni_fail_type),
1610                        event->ni_fail_type,
1611                        ptllnd_evtype2str(event->type), event->type,
1612                        unlinked, ptllnd_msgtype2str(tx->tx_type), tx->tx_type,
1613                        libcfs_id2str(tx->tx_peer->plp_id));
1614
1615         LASSERT (!PtlHandleIsEqual(event->md_handle, PTL_INVALID_HANDLE));
1616
1617         isreq = PtlHandleIsEqual(event->md_handle, tx->tx_reqmdh);
1618         if (isreq) {
1619                 LASSERT (event->md.start == (void *)&tx->tx_msg);
1620                 if (unlinked) {
1621                         tx->tx_reqmdh = PTL_INVALID_HANDLE;
1622                         gettimeofday(&tx->tx_req_done, NULL);
1623                 }
1624         }
1625
1626         isbulk = PtlHandleIsEqual(event->md_handle, tx->tx_bulkmdh);
1627         if ( isbulk && unlinked ) {
1628                 tx->tx_bulkmdh = PTL_INVALID_HANDLE;
1629                 gettimeofday(&tx->tx_bulk_done, NULL);
1630         }
1631
1632         LASSERT (!isreq != !isbulk);            /* always one and only 1 match */
1633
1634         PTLLND_HISTORY("%s[%d/%d+%d(%d)]: TX done %p %s%s",
1635                        libcfs_id2str(tx->tx_peer->plp_id), 
1636                        tx->tx_peer->plp_credits,
1637                        tx->tx_peer->plp_outstanding_credits,
1638                        tx->tx_peer->plp_sent_credits,
1639                        plni->plni_peer_credits + tx->tx_peer->plp_lazy_credits,
1640                        tx, isreq ? "REQ" : "BULK", unlinked ? "(unlinked)" : "");
1641
1642         LASSERT (!isreq != !isbulk);            /* always one and only 1 match */
1643         switch (tx->tx_type) {
1644         default:
1645                 LBUG();
1646
1647         case PTLLND_MSG_TYPE_NOOP:
1648         case PTLLND_MSG_TYPE_HELLO:
1649         case PTLLND_MSG_TYPE_IMMEDIATE:
1650                 LASSERT (event->type == PTL_EVENT_UNLINK ||
1651                          event->type == PTL_EVENT_SEND_END);
1652                 LASSERT (isreq);
1653                 break;
1654
1655         case PTLLND_MSG_TYPE_GET:
1656                 LASSERT (event->type == PTL_EVENT_UNLINK ||
1657                          (isreq && event->type == PTL_EVENT_SEND_END) ||
1658                          (isbulk && event->type == PTL_EVENT_PUT_END));
1659
1660                 if (isbulk && !error && event->type == PTL_EVENT_PUT_END) {
1661                         /* Check GET matched */
1662                         if (event->hdr_data == PTLLND_RDMA_OK) {
1663                                 lnet_set_reply_msg_len(ni, 
1664                                                        tx->tx_lnetreplymsg,
1665                                                        event->mlength);
1666                         } else {
1667                                 CERROR ("Unmatched GET with %s\n",
1668                                         libcfs_id2str(tx->tx_peer->plp_id));
1669                                 tx->tx_status = -EIO;
1670                         }
1671                 }
1672                 break;
1673
1674         case PTLLND_MSG_TYPE_PUT:
1675                 LASSERT (event->type == PTL_EVENT_UNLINK ||
1676                          (isreq && event->type == PTL_EVENT_SEND_END) ||
1677                          (isbulk && event->type == PTL_EVENT_GET_END));
1678                 break;
1679
1680         case PTLLND_RDMA_READ:
1681                 LASSERT (event->type == PTL_EVENT_UNLINK ||
1682                          event->type == PTL_EVENT_SEND_END ||
1683                          event->type == PTL_EVENT_REPLY_END);
1684                 LASSERT (isbulk);
1685                 break;
1686
1687         case PTLLND_RDMA_WRITE:
1688                 LASSERT (event->type == PTL_EVENT_UNLINK ||
1689                          event->type == PTL_EVENT_SEND_END);
1690                 LASSERT (isbulk);
1691         }
1692
1693         /* Schedule ptllnd_tx_done() on error or last completion event */
1694         if (error ||
1695             (PtlHandleIsEqual(tx->tx_bulkmdh, PTL_INVALID_HANDLE) &&
1696              PtlHandleIsEqual(tx->tx_reqmdh, PTL_INVALID_HANDLE))) {
1697                 if (error)
1698                         tx->tx_status = -EIO;
1699                 list_del(&tx->tx_list);
1700                 list_add_tail(&tx->tx_list, &plni->plni_zombie_txs);
1701         }
1702 }
1703
1704 ptllnd_tx_t *
1705 ptllnd_find_timed_out_tx(ptllnd_peer_t *peer)
1706 {
1707         time_t            now = cfs_time_current_sec();
1708         struct list_head *tmp;
1709
1710         list_for_each(tmp, &peer->plp_txq) {
1711                 ptllnd_tx_t *tx = list_entry(tmp, ptllnd_tx_t, tx_list);
1712                 
1713                 if (tx->tx_deadline < now)
1714                         return tx;
1715         }
1716         
1717         list_for_each(tmp, &peer->plp_activeq) {
1718                 ptllnd_tx_t *tx = list_entry(tmp, ptllnd_tx_t, tx_list);
1719                 
1720                 if (tx->tx_deadline < now)
1721                         return tx;
1722         }
1723
1724         return NULL;
1725 }
1726
1727 void
1728 ptllnd_check_peer(ptllnd_peer_t *peer)
1729 {
1730         ptllnd_tx_t *tx = ptllnd_find_timed_out_tx(peer);
1731         
1732         if (tx == NULL)
1733                 return;
1734         
1735         CERROR("%s: timed out\n", libcfs_id2str(peer->plp_id));
1736         ptllnd_close_peer(peer, -ETIMEDOUT);
1737 }
1738
1739 void
1740 ptllnd_watchdog (lnet_ni_t *ni, time_t now)
1741 {
1742         ptllnd_ni_t      *plni = ni->ni_data;
1743         const int         n = 4;
1744         int               p = plni->plni_watchdog_interval;
1745         int               chunk = plni->plni_peer_hash_size;
1746         int               interval = now - (plni->plni_watchdog_nextt - p);
1747         int               i;
1748         struct list_head *hashlist;
1749         struct list_head *tmp;
1750         struct list_head *nxt;
1751
1752         /* Time to check for RDMA timeouts on a few more peers: 
1753          * I try to do checks every 'p' seconds on a proportion of the peer
1754          * table and I need to check every connection 'n' times within a
1755          * timeout interval, to ensure I detect a timeout on any connection
1756          * within (n+1)/n times the timeout interval. */
1757
1758         LASSERT (now >= plni->plni_watchdog_nextt);
1759
1760         if (plni->plni_timeout > n * interval) { /* Scan less than the whole table? */
1761                 chunk = (chunk * n * interval) / plni->plni_timeout;
1762                 if (chunk == 0)
1763                         chunk = 1;
1764         }
1765
1766         for (i = 0; i < chunk; i++) {
1767                 hashlist = &plni->plni_peer_hash[plni->plni_watchdog_peeridx];
1768                 
1769                 list_for_each_safe(tmp, nxt, hashlist) {
1770                         ptllnd_check_peer(list_entry(tmp, ptllnd_peer_t, plp_list));
1771                 }
1772                 
1773                 plni->plni_watchdog_peeridx = (plni->plni_watchdog_peeridx + 1) %
1774                                               plni->plni_peer_hash_size;
1775         }
1776
1777         plni->plni_watchdog_nextt = now + p;
1778 }
1779
1780 void
1781 ptllnd_wait (lnet_ni_t *ni, int milliseconds)
1782 {
1783         static struct timeval  prevt;
1784         static int             prevt_count;
1785         static int             call_count;
1786
1787         struct timeval         start;
1788         struct timeval         then;
1789         struct timeval         now;
1790         struct timeval         deadline;
1791         
1792         ptllnd_ni_t   *plni = ni->ni_data;
1793         ptllnd_tx_t   *tx;
1794         ptl_event_t    event;
1795         int            which;
1796         int            rc;
1797         int            found = 0;
1798         int            timeout = 0;
1799
1800         /* Handle any currently queued events, returning immediately if any.
1801          * Otherwise block for the timeout and handle all events queued
1802          * then. */
1803
1804         gettimeofday(&start, NULL);
1805         call_count++;
1806
1807         if (milliseconds <= 0) {
1808                 deadline = start;
1809         } else {
1810                 deadline.tv_sec  = start.tv_sec  +  milliseconds/1000;
1811                 deadline.tv_usec = start.tv_usec + (milliseconds % 1000)*1000;
1812
1813                 if (deadline.tv_usec >= 1000000) {
1814                         start.tv_usec -= 1000000;
1815                         start.tv_sec++;
1816                 }
1817         }
1818
1819         for (;;) {
1820                 gettimeofday(&then, NULL);
1821                 
1822                 rc = PtlEQPoll(&plni->plni_eqh, 1, timeout, &event, &which);
1823
1824                 gettimeofday(&now, NULL);
1825
1826                 if ((now.tv_sec*1000 + now.tv_usec/1000) - 
1827                     (then.tv_sec*1000 + then.tv_usec/1000) > timeout + 1000) {
1828                         /* 1000 mS grace...........................^ */
1829                         CERROR("SLOW PtlEQPoll(%d): %dmS elapsed\n", timeout,
1830                                (int)(now.tv_sec*1000 + now.tv_usec/1000) - 
1831                                (int)(then.tv_sec*1000 + then.tv_usec/1000));
1832                 }
1833
1834                 if (rc == PTL_EQ_EMPTY) {
1835                         if (found)              /* handled some events */
1836                                 break;
1837
1838                         if (now.tv_sec >= plni->plni_watchdog_nextt) { /* check timeouts? */
1839                                 ptllnd_watchdog(ni, now.tv_sec);
1840                                 LASSERT (now.tv_sec < plni->plni_watchdog_nextt);
1841                         }
1842                         
1843                         if (now.tv_sec > deadline.tv_sec || /* timeout expired */
1844                             (now.tv_sec == deadline.tv_sec &&
1845                              now.tv_usec >= deadline.tv_usec))
1846                                 break;
1847
1848                         if (milliseconds < 0 ||
1849                             plni->plni_watchdog_nextt <= deadline.tv_sec)  {
1850                                 timeout = (plni->plni_watchdog_nextt - now.tv_sec)*1000;
1851                         } else {
1852                                 timeout = (deadline.tv_sec - now.tv_sec)*1000 +
1853                                           (deadline.tv_usec - now.tv_usec)/1000;
1854                         }
1855
1856                         continue;
1857                 }
1858                 
1859                 LASSERT (rc == PTL_OK || rc == PTL_EQ_DROPPED);
1860
1861                 if (rc == PTL_EQ_DROPPED)
1862                         CERROR("Event queue: size %d is too small\n",
1863                                plni->plni_eq_size);
1864
1865                 timeout = 0;
1866                 found = 1;
1867
1868                 switch (ptllnd_eventarg2type(event.md.user_ptr)) {
1869                 default:
1870                         LBUG();
1871
1872                 case PTLLND_EVENTARG_TYPE_TX:
1873                         ptllnd_tx_event(ni, &event);
1874                         break;
1875
1876                 case PTLLND_EVENTARG_TYPE_BUF:
1877                         ptllnd_buf_event(ni, &event);
1878                         break;
1879                 }
1880         }
1881
1882         while (!list_empty(&plni->plni_zombie_txs)) {
1883                 tx = list_entry(plni->plni_zombie_txs.next,
1884                                 ptllnd_tx_t, tx_list);
1885                 list_del_init(&tx->tx_list);
1886                 ptllnd_tx_done(tx);
1887         }
1888
1889         if (prevt.tv_sec == 0 ||
1890             prevt.tv_sec != now.tv_sec) {
1891                 PTLLND_HISTORY("%d wait entered at %d.%06d - prev %d %d.%06d", 
1892                                call_count, (int)start.tv_sec, (int)start.tv_usec,
1893                                prevt_count, (int)prevt.tv_sec, (int)prevt.tv_usec);
1894                 prevt = now;
1895         }
1896 }