Whamcloud - gitweb
b=10778,i=eeb:
[fs/lustre-release.git] / lnet / klnds / ptllnd / ptllnd_peer.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: PJ Kirner <pjkirner@clusterfs.com>
6  *           E Barton <eeb@bartonsoftware.com>
7  *
8  *   This file is part of the Lustre file system, http://www.lustre.org
9  *   Lustre is a trademark of Cluster File Systems, Inc.
10  *
11  *   This file is confidential source code owned by Cluster File Systems.
12  *   No viewing, modification, compilation, redistribution, or any other
13  *   form of use is permitted except through a signed license agreement.
14  *
15  *   If you have not signed such an agreement, then you have no rights to
16  *   this file.  Please destroy it immediately and contact CFS.
17  *
18  */
19
20 #include "ptllnd.h"
21 #include <libcfs/list.h>
22
23 static int
24 kptllnd_count_queue(struct list_head *q)
25 {
26         struct list_head *e;
27         int               n = 0;
28         
29         list_for_each(e, q) {
30                 n++;
31         }
32
33         return n;
34 }
35
36 int
37 kptllnd_get_peer_info(int index, 
38                       lnet_process_id_t *id,
39                       int *state, int *sent_hello,
40                       int *refcount, __u64 *incarnation,
41                       __u64 *next_matchbits, __u64 *last_matchbits_seen,
42                       int *nsendq, int *nactiveq,
43                       int *credits, int *outstanding_credits) 
44 {
45         rwlock_t         *g_lock = &kptllnd_data.kptl_peer_rw_lock;
46         unsigned long     flags;
47         struct list_head *ptmp;
48         kptl_peer_t      *peer;
49         int               i;
50         int               rc = -ENOENT;
51
52         read_lock_irqsave(g_lock, flags);
53
54         for (i = 0; i < kptllnd_data.kptl_peer_hash_size; i++) {
55                 
56                 list_for_each (ptmp, &kptllnd_data.kptl_peers[i]) {
57                         peer = list_entry(ptmp, kptl_peer_t, peer_list);
58
59                         if (index-- > 0)
60                                 continue;
61                         
62                         *id          = peer->peer_id;
63                         *state       = peer->peer_state;
64                         *sent_hello  = peer->peer_sent_hello;
65                         *refcount    = atomic_read(&peer->peer_refcount);
66                         *incarnation = peer->peer_incarnation;
67
68                         spin_lock(&peer->peer_lock);
69
70                         *next_matchbits      = peer->peer_next_matchbits;
71                         *last_matchbits_seen = peer->peer_last_matchbits_seen;
72                         *credits             = peer->peer_credits;
73                         *outstanding_credits = peer->peer_outstanding_credits;
74
75                         *nsendq   = kptllnd_count_queue(&peer->peer_sendq);
76                         *nactiveq = kptllnd_count_queue(&peer->peer_activeq);
77
78                         spin_unlock(&peer->peer_lock);
79
80                         rc = 0;
81                         goto out;
82                 }
83         }
84         
85  out:
86         read_unlock_irqrestore(g_lock, flags);
87         return rc;
88 }
89
90 void
91 kptllnd_peer_add_peertable_locked (kptl_peer_t *peer)
92 {
93         LASSERT (!kptllnd_data.kptl_shutdown);
94         LASSERT (kptllnd_data.kptl_n_active_peers <
95                  kptllnd_data.kptl_expected_peers);
96
97         LASSERT (peer->peer_state == PEER_STATE_WAITING_HELLO ||
98                  peer->peer_state == PEER_STATE_ACTIVE);
99         
100         kptllnd_data.kptl_n_active_peers++;
101         atomic_inc(&peer->peer_refcount);       /* +1 ref for the list */
102
103         /* NB add to HEAD of peer list for MRU order!
104          * (see kptllnd_cull_peertable) */
105         list_add(&peer->peer_list, kptllnd_nid2peerlist(peer->peer_id.nid));
106 }
107
108 void
109 kptllnd_cull_peertable_locked (lnet_process_id_t pid)
110 {
111         /* I'm about to add a new peer with this portals ID to the peer table,
112          * so (a) this peer should not exist already and (b) I want to leave at
113          * most (max_procs_per_nid - 1) peers with this NID in the table. */
114         struct list_head  *peers = kptllnd_nid2peerlist(pid.nid);
115         int                cull_count = *kptllnd_tunables.kptl_max_procs_per_node;
116         int                count;
117         struct list_head  *tmp;
118         struct list_head  *nxt;
119         kptl_peer_t       *peer;
120         
121         count = 0;
122         list_for_each_safe (tmp, nxt, peers) {
123                 /* NB I rely on kptllnd_peer_add_peertable_locked to add peers
124                  * in MRU order */
125                 peer = list_entry(tmp, kptl_peer_t, peer_list);
126                         
127                 if (peer->peer_id.nid != pid.nid)
128                         continue;
129
130                 LASSERT (peer->peer_id.pid != pid.pid);
131                         
132                 count++;
133
134                 if (count < cull_count) /* recent (don't cull) */
135                         continue;
136
137                 CDEBUG(D_NET, "Cull %s(%s)\n",
138                        libcfs_id2str(peer->peer_id),
139                        kptllnd_ptlid2str(peer->peer_ptlid));
140                 
141                 kptllnd_peer_close_locked(peer, 0);
142         }
143 }
144
145 kptl_peer_t *
146 kptllnd_peer_allocate (lnet_process_id_t lpid, ptl_process_id_t ppid)
147 {
148         unsigned long    flags;
149         kptl_peer_t     *peer;
150
151         LIBCFS_ALLOC(peer, sizeof (*peer));
152         if (peer == NULL) {
153                 CERROR("Can't create peer %s (%s)\n",
154                        libcfs_id2str(lpid), 
155                        kptllnd_ptlid2str(ppid));
156                 return NULL;
157         }
158
159         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
160
161         INIT_LIST_HEAD (&peer->peer_sendq);
162         INIT_LIST_HEAD (&peer->peer_activeq);
163         spin_lock_init (&peer->peer_lock);
164
165         peer->peer_state = PEER_STATE_ALLOCATED;
166         peer->peer_error = 0;
167         peer->peer_last_alive = cfs_time_current();
168         peer->peer_id = lpid;
169         peer->peer_ptlid = ppid;
170         peer->peer_credits = 1;                 /* enough for HELLO */
171         peer->peer_next_matchbits = PTL_RESERVED_MATCHBITS;
172         peer->peer_outstanding_credits = *kptllnd_tunables.kptl_peercredits - 1;
173         peer->peer_sent_credits = 1;           /* HELLO credit is implicit */
174         peer->peer_max_msg_size = PTLLND_MIN_BUFFER_SIZE; /* until we know better */
175
176         atomic_set(&peer->peer_refcount, 1);    /* 1 ref for caller */
177
178         write_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
179
180         peer->peer_myincarnation = kptllnd_data.kptl_incarnation;
181
182         /* Only increase # peers under lock, to guarantee we dont grow it
183          * during shutdown */
184         if (kptllnd_data.kptl_shutdown) {
185                 write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, 
186                                         flags);
187                 LIBCFS_FREE(peer, sizeof(*peer));
188                 return NULL;
189         }
190
191         kptllnd_data.kptl_npeers++;
192         write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
193         
194         return peer;
195 }
196
197 void
198 kptllnd_peer_destroy (kptl_peer_t *peer)
199 {
200         unsigned long flags;
201         
202         CDEBUG(D_NET, "Peer=%p\n", peer);
203
204         LASSERT (!in_interrupt());
205         LASSERT (atomic_read(&peer->peer_refcount) == 0);
206         LASSERT (peer->peer_state == PEER_STATE_ALLOCATED ||
207                  peer->peer_state == PEER_STATE_ZOMBIE);
208         LASSERT (list_empty(&peer->peer_sendq));
209         LASSERT (list_empty(&peer->peer_activeq));
210
211         write_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
212
213         if (peer->peer_state == PEER_STATE_ZOMBIE)
214                 list_del(&peer->peer_list);
215
216         kptllnd_data.kptl_npeers--;
217
218         write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
219
220         LIBCFS_FREE (peer, sizeof (*peer));
221 }
222
223 void
224 kptllnd_cancel_txlist (struct list_head *peerq, struct list_head *txs)
225 {
226         struct list_head  *tmp;
227         struct list_head  *nxt;
228         kptl_tx_t         *tx;
229
230         list_for_each_safe (tmp, nxt, peerq) {
231                 tx = list_entry(tmp, kptl_tx_t, tx_list);
232
233                 list_del(&tx->tx_list);
234                 list_add_tail(&tx->tx_list, txs);
235
236                 tx->tx_status = -EIO;
237                 tx->tx_active = 0;
238         }
239 }
240
241 void
242 kptllnd_peer_cancel_txs(kptl_peer_t *peer, struct list_head *txs)
243 {
244         unsigned long   flags;
245
246         spin_lock_irqsave(&peer->peer_lock, flags);
247
248         kptllnd_cancel_txlist(&peer->peer_sendq, txs);
249         kptllnd_cancel_txlist(&peer->peer_activeq, txs);
250                 
251         spin_unlock_irqrestore(&peer->peer_lock, flags);
252 }
253
254 void
255 kptllnd_peer_alive (kptl_peer_t *peer)
256 {
257         /* This is racy, but everyone's only writing cfs_time_current() */
258         peer->peer_last_alive = cfs_time_current();
259         mb();
260 }
261
262 void
263 kptllnd_peer_notify (kptl_peer_t *peer)
264 {
265         unsigned long flags;
266         time_t        last_alive = 0;
267         int           error = 0;
268         
269         spin_lock_irqsave(&peer->peer_lock, flags);
270
271         if (peer->peer_error != 0) {
272                 error = peer->peer_error;
273                 peer->peer_error = 0;
274                 
275                 last_alive = cfs_time_current_sec() - 
276                              cfs_duration_sec(cfs_time_current() - 
277                                               peer->peer_last_alive);
278         }
279         
280         spin_unlock_irqrestore(&peer->peer_lock, flags);
281
282         if (error != 0)
283                 lnet_notify (kptllnd_data.kptl_ni, peer->peer_id.nid, 0,
284                              last_alive);
285 }
286
287 void
288 kptllnd_handle_closing_peers ()
289 {
290         unsigned long           flags;
291         struct list_head        txs;
292         kptl_peer_t            *peer;
293         struct list_head       *tmp;
294         struct list_head       *nxt;
295         kptl_tx_t              *tx;
296         int                     idle;
297
298         /* Check with a read lock first to avoid blocking anyone */
299
300         read_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
301         idle = list_empty(&kptllnd_data.kptl_closing_peers) &&
302                list_empty(&kptllnd_data.kptl_zombie_peers);
303         read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
304
305         if (idle)
306                 return;
307
308         INIT_LIST_HEAD(&txs);
309
310         write_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
311
312         /* Cancel txs on all zombie peers.  NB anyone dropping the last peer
313          * ref removes it from this list, so I musn't drop the lock while
314          * scanning it. */
315         list_for_each (tmp, &kptllnd_data.kptl_zombie_peers) {
316                 peer = list_entry (tmp, kptl_peer_t, peer_list);
317
318                 LASSERT (peer->peer_state == PEER_STATE_ZOMBIE);
319
320                 kptllnd_peer_cancel_txs(peer, &txs);
321         }
322
323         /* Notify LNET and cancel txs on closing (i.e. newly closed) peers.  NB
324          * I'm the only one removing from this list, but peers can be added on
325          * the end any time I drop the lock. */
326
327         list_for_each_safe (tmp, nxt, &kptllnd_data.kptl_closing_peers) {
328                 peer = list_entry (tmp, kptl_peer_t, peer_list);
329
330                 LASSERT (peer->peer_state == PEER_STATE_CLOSING);
331
332                 list_del(&peer->peer_list);
333                 list_add_tail(&peer->peer_list,
334                               &kptllnd_data.kptl_zombie_peers);
335                 peer->peer_state = PEER_STATE_ZOMBIE;
336
337                 write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
338
339                 kptllnd_peer_notify(peer);
340                 kptllnd_peer_cancel_txs(peer, &txs);
341                 kptllnd_peer_decref(peer);
342
343                 write_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
344         }
345
346         write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
347
348         /* Drop peer's ref on all cancelled txs.  This will get
349          * kptllnd_tx_fini() to abort outstanding comms if necessary. */
350
351         list_for_each_safe (tmp, nxt, &txs) {
352                 tx = list_entry(tmp, kptl_tx_t, tx_list);
353                 list_del(&tx->tx_list);
354                 kptllnd_tx_decref(tx);
355         }
356 }
357
358 void
359 kptllnd_peer_close_locked(kptl_peer_t *peer, int why)
360 {
361         switch (peer->peer_state) {
362         default:
363                 LBUG();
364
365         case PEER_STATE_WAITING_HELLO:
366         case PEER_STATE_ACTIVE:
367                 /* Ensure new peers see a new incarnation of me */
368                 LASSERT(peer->peer_myincarnation <= kptllnd_data.kptl_incarnation);
369                 if (peer->peer_myincarnation == kptllnd_data.kptl_incarnation)
370                         kptllnd_data.kptl_incarnation++;
371
372                 /* Removing from peer table */
373                 kptllnd_data.kptl_n_active_peers--;
374                 LASSERT (kptllnd_data.kptl_n_active_peers >= 0);
375
376                 list_del(&peer->peer_list);
377                 kptllnd_peer_unreserve_buffers();
378
379                 peer->peer_error = why; /* stash 'why' only on first close */
380                 peer->peer_state = PEER_STATE_CLOSING;
381
382                 /* Schedule for immediate attention, taking peer table's ref */
383                 list_add_tail(&peer->peer_list, 
384                               &kptllnd_data.kptl_closing_peers);
385                 wake_up(&kptllnd_data.kptl_watchdog_waitq);
386                 break;
387
388         case PEER_STATE_ZOMBIE:
389         case PEER_STATE_CLOSING:
390                 break;
391         }
392 }
393
394 void
395 kptllnd_peer_close(kptl_peer_t *peer, int why)
396 {
397         unsigned long      flags;
398
399         write_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
400         kptllnd_peer_close_locked(peer, why);
401         write_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
402 }
403
404 int
405 kptllnd_peer_del(lnet_process_id_t id)
406 {
407         struct list_head  *ptmp;
408         struct list_head  *pnxt;
409         kptl_peer_t       *peer;
410         int                lo;
411         int                hi;
412         int                i;
413         unsigned long      flags;
414         int                rc = -ENOENT;
415
416         /*
417          * Find the single bucket we are supposed to look at or if nid is a
418          * wildcard (LNET_NID_ANY) then look at all of the buckets
419          */
420         if (id.nid != LNET_NID_ANY) {
421                 struct list_head *l = kptllnd_nid2peerlist(id.nid);
422                 
423                 lo = hi =  l - kptllnd_data.kptl_peers;
424         } else {
425                 if (id.pid != LNET_PID_ANY)
426                         return -EINVAL;
427                 
428                 lo = 0;
429                 hi = kptllnd_data.kptl_peer_hash_size - 1;
430         }
431
432 again:
433         read_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
434
435         for (i = lo; i <= hi; i++) {
436                 list_for_each_safe (ptmp, pnxt, &kptllnd_data.kptl_peers[i]) {
437                         peer = list_entry (ptmp, kptl_peer_t, peer_list);
438
439                         if (!(id.nid == LNET_NID_ANY || 
440                               (peer->peer_id.nid == id.nid &&
441                                (id.pid == LNET_PID_ANY || 
442                                 peer->peer_id.pid == id.pid))))
443                                 continue;
444
445                         kptllnd_peer_addref(peer); /* 1 ref for me... */
446
447                         read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock,
448                                                flags);
449
450                         kptllnd_peer_close(peer, 0);
451                         kptllnd_peer_decref(peer); /* ...until here */
452
453                         rc = 0;         /* matched something */
454
455                         /* start again now I've dropped the lock */
456                         goto again;
457                 }
458         }
459
460         read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
461
462         return (rc);
463 }
464
465 void
466 kptllnd_post_tx(kptl_peer_t *peer, kptl_tx_t *tx, int nfrag)
467 {
468         /* CAVEAT EMPTOR: I take over caller's ref on 'tx' */
469         ptl_handle_md_t  msg_mdh;
470         ptl_md_t         md;
471         ptl_err_t        prc;
472         unsigned long    flags;
473
474         LASSERT (!tx->tx_idle);
475         LASSERT (!tx->tx_active);
476         LASSERT (PtlHandleIsEqual(tx->tx_msg_mdh, PTL_INVALID_HANDLE));
477         LASSERT (PtlHandleIsEqual(tx->tx_rdma_mdh, PTL_INVALID_HANDLE));
478         LASSERT (tx->tx_type == TX_TYPE_SMALL_MESSAGE ||
479                  tx->tx_type == TX_TYPE_PUT_REQUEST ||
480                  tx->tx_type == TX_TYPE_GET_REQUEST);
481
482         kptllnd_set_tx_peer(tx, peer);
483
484         memset(&md, 0, sizeof(md));
485
486         md.threshold = tx->tx_acked ? 2 : 1;    /* SEND END + ACK? */
487         md.options = PTL_MD_OP_PUT |
488                      PTL_MD_LUSTRE_COMPLETION_SEMANTICS |
489                      PTL_MD_EVENT_START_DISABLE;
490         md.user_ptr = &tx->tx_msg_eventarg;
491         md.eq_handle = kptllnd_data.kptl_eqh;
492
493         if (nfrag == 0) {
494                 md.start = tx->tx_msg;
495                 md.length = tx->tx_msg->ptlm_nob;
496         } else {
497                 LASSERT (nfrag > 1);
498                 LASSERT (tx->tx_frags->iov[0].iov_base == (void *)tx->tx_msg);
499
500                 md.start = tx->tx_frags;
501                 md.length = nfrag;
502                 md.options |= PTL_MD_IOVEC;
503         }
504
505         prc = PtlMDBind(kptllnd_data.kptl_nih, md, PTL_UNLINK, &msg_mdh);
506         if (prc != PTL_OK) {
507                 CERROR("PtlMDBind(%s) failed: %s(%d)\n",
508                        libcfs_id2str(peer->peer_id),
509                        kptllnd_errtype2str(prc), prc);
510                 tx->tx_status = -EIO;
511                 kptllnd_tx_decref(tx);
512                 return;
513         }
514         
515         spin_lock_irqsave(&peer->peer_lock, flags);
516
517         tx->tx_deadline = jiffies + (*kptllnd_tunables.kptl_timeout * HZ);
518         tx->tx_active = 1;
519         tx->tx_msg_mdh = msg_mdh;
520
521         /* Ensure HELLO is sent first */
522         if (tx->tx_msg->ptlm_type == PTLLND_MSG_TYPE_HELLO)
523                 list_add(&tx->tx_list, &peer->peer_sendq);
524         else
525                 list_add_tail(&tx->tx_list, &peer->peer_sendq);
526
527         spin_unlock_irqrestore(&peer->peer_lock, flags);
528 }
529
530 void
531 kptllnd_peer_check_sends (kptl_peer_t *peer)
532 {
533         ptl_handle_me_t  meh;
534         kptl_tx_t       *tx;
535         int              rc;
536         unsigned long    flags;
537
538         LASSERT(!in_interrupt());
539
540         spin_lock_irqsave(&peer->peer_lock, flags);
541
542         peer->peer_retry_noop = 0;
543
544         if (list_empty(&peer->peer_sendq) &&
545             peer->peer_outstanding_credits >= PTLLND_CREDIT_HIGHWATER &&
546             peer->peer_credits != 0) {
547
548                 /* post a NOOP to return credits */
549                 spin_unlock_irqrestore(&peer->peer_lock, flags);
550
551                 tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
552                 if (tx == NULL) {
553                         CERROR("Can't return credits to %s: can't allocate descriptor\n",
554                                libcfs_id2str(peer->peer_id));
555                 } else {
556                         kptllnd_init_msg(tx->tx_msg, PTLLND_MSG_TYPE_NOOP, 0);
557                         kptllnd_post_tx(peer, tx, 0);
558                 }
559
560                 spin_lock_irqsave(&peer->peer_lock, flags);
561                 peer->peer_retry_noop = (tx == NULL);
562         }
563
564         while (!list_empty(&peer->peer_sendq)) {
565                 tx = list_entry (peer->peer_sendq.next, kptl_tx_t, tx_list);
566
567                 LASSERT (tx->tx_active);
568                 LASSERT (!PtlHandleIsEqual(tx->tx_msg_mdh, PTL_INVALID_HANDLE));
569                 LASSERT (PtlHandleIsEqual(tx->tx_rdma_mdh, PTL_INVALID_HANDLE));
570
571                 LASSERT (peer->peer_outstanding_credits >= 0);
572                 LASSERT (peer->peer_sent_credits >= 0);
573                 LASSERT (peer->peer_sent_credits +
574                          peer->peer_outstanding_credits <=
575                          *kptllnd_tunables.kptl_peercredits);
576                 LASSERT (peer->peer_credits >= 0);
577
578                 /* Ensure HELLO is sent first */
579                 if (!peer->peer_sent_hello) {
580                         if (tx->tx_msg->ptlm_type != PTLLND_MSG_TYPE_HELLO)
581                                 break;
582                         peer->peer_sent_hello = 1;
583                 }
584
585                 if (peer->peer_credits == 0) {
586                         CDEBUG(D_NETTRACE, "%s[%d/%d+%d]: no credits for %p\n",
587                                libcfs_id2str(peer->peer_id), 
588                                peer->peer_credits,
589                                peer->peer_outstanding_credits, 
590                                peer->peer_sent_credits, tx);
591                         break;
592                 }
593
594                 /* Don't use the last credit unless I've got credits to
595                  * return */
596                 if (peer->peer_credits == 1 &&
597                     peer->peer_outstanding_credits == 0) {
598                         CDEBUG(D_NETTRACE, "%s[%d/%d+%d]: "
599                                "not using last credit for %p\n",
600                                libcfs_id2str(peer->peer_id), 
601                                peer->peer_credits,
602                                peer->peer_outstanding_credits,
603                                peer->peer_sent_credits, tx);
604                         break;
605                 }
606
607                 list_del(&tx->tx_list);
608
609                 /* Discard any NOOP I queued if I'm not at the high-water mark
610                  * any more or more messages have been queued */
611                 if (tx->tx_msg->ptlm_type == PTLLND_MSG_TYPE_NOOP &&
612                     (!list_empty(&peer->peer_sendq) ||
613                      peer->peer_outstanding_credits < PTLLND_CREDIT_HIGHWATER)) {
614
615                         tx->tx_active = 0;
616
617                         spin_unlock_irqrestore(&peer->peer_lock, flags);
618
619                         CDEBUG(D_NET, "%s: redundant noop\n", 
620                                libcfs_id2str(peer->peer_id));
621                         kptllnd_tx_decref(tx);
622
623                         spin_lock_irqsave(&peer->peer_lock, flags);
624                         continue;
625                 }
626
627                 /* fill last-minute msg fields */
628                 kptllnd_msg_pack(tx->tx_msg, peer);
629
630                 if (tx->tx_type == TX_TYPE_PUT_REQUEST ||
631                     tx->tx_type == TX_TYPE_GET_REQUEST) {
632                         /* peer_next_matchbits must be known good */
633                         LASSERT (peer->peer_state >= PEER_STATE_ACTIVE);
634                         /* Assume 64-bit matchbits can't wrap */
635                         LASSERT (peer->peer_next_matchbits >= PTL_RESERVED_MATCHBITS);
636                         tx->tx_msg->ptlm_u.rdma.kptlrm_matchbits =
637                                 peer->peer_next_matchbits++;
638                 }
639                 
640                 peer->peer_sent_credits += peer->peer_outstanding_credits;
641                 peer->peer_outstanding_credits = 0;
642                 peer->peer_credits--;
643
644                 CDEBUG(D_NETTRACE, "%s[%d/%d+%d]: %s tx=%p nob=%d cred=%d\n",
645                        libcfs_id2str(peer->peer_id), peer->peer_credits,
646                        peer->peer_outstanding_credits, peer->peer_sent_credits,
647                        kptllnd_msgtype2str(tx->tx_msg->ptlm_type),
648                        tx, tx->tx_msg->ptlm_nob,
649                        tx->tx_msg->ptlm_credits);
650
651                 list_add_tail(&tx->tx_list, &peer->peer_activeq);
652
653                 kptllnd_tx_addref(tx);          /* 1 ref for me... */
654
655                 spin_unlock_irqrestore(&peer->peer_lock, flags);
656
657                 if (tx->tx_type == TX_TYPE_PUT_REQUEST ||
658                     tx->tx_type == TX_TYPE_GET_REQUEST) {
659                         /* Post bulk now we have safe matchbits */
660                         rc = PtlMEAttach(kptllnd_data.kptl_nih,
661                                          *kptllnd_tunables.kptl_portal,
662                                          peer->peer_ptlid,
663                                          tx->tx_msg->ptlm_u.rdma.kptlrm_matchbits,
664                                          0,             /* ignore bits */
665                                          PTL_UNLINK,
666                                          PTL_INS_BEFORE,
667                                          &meh);
668                         if (rc != PTL_OK) {
669                                 CERROR("PtlMEAttach(%s) failed: %s(%d)\n",
670                                        libcfs_id2str(peer->peer_id),
671                                        kptllnd_errtype2str(rc), rc);
672                                 goto failed;
673                         }
674
675                         rc = PtlMDAttach(meh, tx->tx_rdma_md, PTL_UNLINK,
676                                          &tx->tx_rdma_mdh);
677                         if (rc != PTL_OK) {
678                                 CERROR("PtlMDAttach(%s) failed: %s(%d)\n",
679                                        libcfs_id2str(tx->tx_peer->peer_id),
680                                        kptllnd_errtype2str(rc), rc);
681                                 rc = PtlMEUnlink(meh);
682                                 LASSERT(rc == PTL_OK);
683                                 tx->tx_rdma_mdh = PTL_INVALID_HANDLE;
684                                 goto failed;
685                         }
686                         /* I'm not racing with the event callback here.  It's a
687                          * bug if there's an event on the MD I just attached
688                          * before I actually send the RDMA request message -
689                          * probably matchbits re-used in error. */
690                 }
691
692                 tx->tx_tposted = jiffies;       /* going on the wire */
693
694                 rc = PtlPut (tx->tx_msg_mdh,
695                              tx->tx_acked ? PTL_ACK_REQ : PTL_NOACK_REQ,
696                              peer->peer_ptlid,
697                              *kptllnd_tunables.kptl_portal,
698                              0,                 /* acl cookie */
699                              LNET_MSG_MATCHBITS,
700                              0,                 /* offset */
701                              0);                /* header data */
702                 if (rc != PTL_OK) {
703                         CERROR("PtlPut %s error %s(%d)\n",
704                                libcfs_id2str(peer->peer_id),
705                                kptllnd_errtype2str(rc), rc);
706                         goto failed;
707                 }
708
709                 kptllnd_tx_decref(tx);          /* drop my ref */
710
711                 spin_lock_irqsave(&peer->peer_lock, flags);
712         }
713
714         spin_unlock_irqrestore(&peer->peer_lock, flags);
715         return;
716
717  failed:
718         /* Nuke everything (including tx we were trying) */
719         kptllnd_peer_close(peer, -EIO);
720         kptllnd_tx_decref(tx);
721 }
722
723 kptl_tx_t *
724 kptllnd_find_timed_out_tx(kptl_peer_t *peer)
725 {
726         kptl_tx_t         *tx;
727         struct list_head  *tmp;
728
729         list_for_each(tmp, &peer->peer_sendq) {
730                 tx = list_entry(peer->peer_sendq.next, kptl_tx_t, tx_list);
731
732                 if (time_after_eq(jiffies, tx->tx_deadline)) {
733                         kptllnd_tx_addref(tx);
734                         return tx;
735                 }
736         }
737
738         list_for_each(tmp, &peer->peer_activeq) {
739                 tx = list_entry(peer->peer_activeq.next, kptl_tx_t, tx_list);
740
741                 if (time_after_eq(jiffies, tx->tx_deadline)) {
742                         kptllnd_tx_addref(tx);
743                         return tx;
744                 }
745         }
746
747         return NULL;
748 }
749
750
751 void
752 kptllnd_peer_check_bucket (int idx, int stamp)
753 {
754         struct list_head  *peers = &kptllnd_data.kptl_peers[idx];
755         struct list_head  *ptmp;
756         kptl_peer_t       *peer;
757         kptl_tx_t         *tx;
758         unsigned long      flags;
759         int                nsend;
760         int                nactive;
761         int                check_sends;
762
763         CDEBUG(D_NET, "Bucket=%d, stamp=%d\n", idx, stamp);
764
765  again:
766         /* NB. Shared lock while I just look */
767         read_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
768
769         list_for_each (ptmp, peers) {
770                 peer = list_entry (ptmp, kptl_peer_t, peer_list);
771
772                 CDEBUG(D_NET, "Peer=%s Credits=%d Outstanding=%d Send=%d\n",
773                        libcfs_id2str(peer->peer_id), peer->peer_credits, 
774                        peer->peer_outstanding_credits, peer->peer_sent_credits);
775
776                 spin_lock(&peer->peer_lock);
777
778                 if (peer->peer_check_stamp == stamp) {
779                         /* checked already this pass */
780                         spin_unlock(&peer->peer_lock);
781                         continue;
782                 }
783
784                 peer->peer_check_stamp = stamp;
785                 tx = kptllnd_find_timed_out_tx(peer);
786                 check_sends = peer->peer_retry_noop;
787                 
788                 spin_unlock(&peer->peer_lock);
789                 
790                 if (tx == NULL && !check_sends)
791                         continue;
792
793                 kptllnd_peer_addref(peer); /* 1 ref for me... */
794
795                 read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
796
797                 if (tx == NULL) { /* nothing timed out */
798                         kptllnd_peer_check_sends(peer);
799                         kptllnd_peer_decref(peer); /* ...until here or... */
800
801                         /* rescan after dropping the lock */
802                         goto again;
803                 }
804
805                 spin_lock_irqsave(&peer->peer_lock, flags);
806                 nsend = kptllnd_count_queue(&peer->peer_sendq);
807                 nactive = kptllnd_count_queue(&peer->peer_activeq);
808                 spin_unlock_irqrestore(&peer->peer_lock, flags);
809
810                 LCONSOLE_ERROR_MSG(0x126, "Timing out %s: %s\n",
811                                    libcfs_id2str(peer->peer_id),
812                                    (tx->tx_tposted == 0) ? 
813                                    "no free peer buffers" : 
814                                    "please check Portals");
815
816                 CERROR("%s timed out: cred %d outstanding %d, sent %d, "
817                        "sendq %d, activeq %d Tx %p %s (%s%s%s) status %d "
818                        "%sposted %lu T/O %ds\n",
819                        libcfs_id2str(peer->peer_id), peer->peer_credits,
820                        peer->peer_outstanding_credits, peer->peer_sent_credits,
821                        nsend, nactive, tx, kptllnd_tx_typestr(tx->tx_type),
822                        tx->tx_active ? "A" : "",
823                        PtlHandleIsEqual(tx->tx_msg_mdh, PTL_INVALID_HANDLE) ?
824                        "" : "M",
825                        PtlHandleIsEqual(tx->tx_rdma_mdh, PTL_INVALID_HANDLE) ?
826                        "" : "D",
827                        tx->tx_status,
828                        (tx->tx_tposted == 0) ? "not " : "",
829                        (tx->tx_tposted == 0) ? 0UL : (jiffies - tx->tx_tposted),
830                        *kptllnd_tunables.kptl_timeout);
831
832                 kptllnd_dump_ptltrace();
833
834                 kptllnd_tx_decref(tx);
835
836                 kptllnd_peer_close(peer, -ETIMEDOUT);
837                 kptllnd_peer_decref(peer); /* ...until here */
838
839                 /* start again now I've dropped the lock */
840                 goto again;
841         }
842
843         read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);
844 }
845
846 kptl_peer_t *
847 kptllnd_id2peer_locked (lnet_process_id_t id)
848 {
849         struct list_head *peers = kptllnd_nid2peerlist(id.nid);
850         struct list_head *tmp;
851         kptl_peer_t      *peer;
852
853         list_for_each (tmp, peers) {
854
855                 peer = list_entry (tmp, kptl_peer_t, peer_list);
856
857                 LASSERT(peer->peer_state == PEER_STATE_WAITING_HELLO ||
858                         peer->peer_state == PEER_STATE_ACTIVE);
859                 
860                 if (peer->peer_id.nid != id.nid ||
861                     peer->peer_id.pid != id.pid)
862                         continue;
863
864                 kptllnd_peer_addref(peer);
865
866                 CDEBUG(D_NET, "%s -> %s (%d)\n",
867                        libcfs_id2str(id), 
868                        kptllnd_ptlid2str(peer->peer_ptlid),
869                        atomic_read (&peer->peer_refcount));
870                 return peer;
871         }
872
873         return NULL;
874 }
875
876 void
877 kptllnd_peertable_overflow_msg(char *str, lnet_process_id_t id)
878 {
879         LCONSOLE_ERROR_MSG(0x127, "%s %s overflows the peer table[%d]: "
880                            "messages may be dropped\n",
881                            str, libcfs_id2str(id),
882                            kptllnd_data.kptl_n_active_peers);
883         LCONSOLE_ERROR_MSG(0x128, "Please correct by increasing "
884                            "'max_nodes' or 'max_procs_per_node'\n");
885 }
886
887 __u64
888 kptllnd_get_last_seen_matchbits_locked(lnet_process_id_t lpid)
889 {
890         kptl_peer_t            *peer;
891         struct list_head       *tmp;
892
893         /* Find the last matchbits I saw this new peer using.  Note..
894            A. This peer cannot be in the peer table - she's new!
895            B. If I can't find the peer in the closing/zombie peers, all
896               matchbits are safe because all refs to the (old) peer have gone
897               so all txs have completed so there's no risk of matchbit
898               collision!
899          */
900
901         LASSERT(kptllnd_id2peer_locked(lpid) == NULL);
902
903         /* peer's last matchbits can't change after it comes out of the peer
904          * table, so first match is fine */
905
906         list_for_each (tmp, &kptllnd_data.kptl_closing_peers) {
907                 peer = list_entry (tmp, kptl_peer_t, peer_list);
908
909                 if (peer->peer_id.nid == lpid.nid &&
910                     peer->peer_id.pid == lpid.pid)
911                         return peer->peer_last_matchbits_seen;
912         }
913         
914         list_for_each (tmp, &kptllnd_data.kptl_zombie_peers) {
915                 peer = list_entry (tmp, kptl_peer_t, peer_list);
916
917                 if (peer->peer_id.nid == lpid.nid &&
918                     peer->peer_id.pid == lpid.pid)
919                         return peer->peer_last_matchbits_seen;
920         }
921         
922         return PTL_RESERVED_MATCHBITS;
923 }
924
925 kptl_peer_t *
926 kptllnd_peer_handle_hello (ptl_process_id_t  initiator,
927                            kptl_msg_t       *msg)
928 {
929         rwlock_t           *g_lock = &kptllnd_data.kptl_peer_rw_lock;
930         kptl_peer_t        *peer;
931         kptl_peer_t        *new_peer;
932         lnet_process_id_t   lpid;
933         unsigned long       flags;
934         kptl_tx_t          *hello_tx;
935         int                 rc;
936         __u64               safe_matchbits;
937         __u64               last_matchbits_seen;
938
939         lpid.nid = msg->ptlm_srcnid;
940         lpid.pid = msg->ptlm_srcpid;
941
942         CDEBUG(D_NET, "hello from %s(%s)\n",
943                libcfs_id2str(lpid), kptllnd_ptlid2str(initiator));
944
945         if (initiator.pid != kptllnd_data.kptl_portals_id.pid &&
946             (msg->ptlm_srcpid & LNET_PID_USERFLAG) == 0) {
947                 /* If the peer's PID isn't _the_ ptllnd kernel pid, she must be
948                  * userspace.  Refuse the connection if she hasn't set the
949                  * correct flag in her PID... */
950                 CERROR("Userflag not set in hello from %s (%s)\n",
951                        libcfs_id2str(lpid), kptllnd_ptlid2str(initiator));
952                 return NULL;
953         }
954         
955         /* kptlhm_matchbits are the highest matchbits my peer may have used to
956          * RDMA to me.  I ensure I never register buffers for RDMA that could
957          * match any she used */
958         safe_matchbits = msg->ptlm_u.hello.kptlhm_matchbits + 1;
959
960         if (safe_matchbits < PTL_RESERVED_MATCHBITS) {
961                 CERROR("Illegal matchbits "LPX64" in HELLO from %s\n",
962                        safe_matchbits, libcfs_id2str(lpid));
963                 return NULL;
964         }
965         
966         if (msg->ptlm_u.hello.kptlhm_max_msg_size < PTLLND_MIN_BUFFER_SIZE) {
967                 CERROR("%s: max message size %d < MIN %d",
968                        libcfs_id2str(lpid),
969                        msg->ptlm_u.hello.kptlhm_max_msg_size,
970                        PTLLND_MIN_BUFFER_SIZE);
971                 return NULL;
972         }
973
974         if (msg->ptlm_credits <= 1) {
975                 CERROR("Need more than 1+%d credits from %s\n",
976                        msg->ptlm_credits, libcfs_id2str(lpid));
977                 return NULL;
978         }
979         
980         write_lock_irqsave(g_lock, flags);
981
982         peer = kptllnd_id2peer_locked(lpid);
983         if (peer != NULL) {
984                 if (peer->peer_state == PEER_STATE_WAITING_HELLO) {
985                         /* Completing HELLO handshake */
986                         LASSERT(peer->peer_incarnation == 0);
987
988                         if (msg->ptlm_dststamp != 0 &&
989                             msg->ptlm_dststamp != peer->peer_myincarnation) {
990                                 write_unlock_irqrestore(g_lock, flags);
991
992                                 CERROR("Ignoring HELLO from %s: unexpected "
993                                        "dststamp "LPX64" ("LPX64" wanted)\n",
994                                        libcfs_id2str(lpid),
995                                        msg->ptlm_dststamp,
996                                        peer->peer_myincarnation);
997                                 kptllnd_peer_decref(peer);
998                                 return NULL;
999                         }
1000                         
1001                         /* Concurrent initiation or response to my HELLO */
1002                         peer->peer_state = PEER_STATE_ACTIVE;
1003                         peer->peer_incarnation = msg->ptlm_srcstamp;
1004                         peer->peer_next_matchbits = safe_matchbits;
1005                         peer->peer_max_msg_size =
1006                                 msg->ptlm_u.hello.kptlhm_max_msg_size;
1007                         
1008                         write_unlock_irqrestore(g_lock, flags);
1009                         return peer;
1010                 }
1011
1012                 if (msg->ptlm_dststamp != 0 &&
1013                     msg->ptlm_dststamp <= peer->peer_myincarnation) {
1014                         write_unlock_irqrestore(g_lock, flags);
1015
1016                         CERROR("Ignoring stale HELLO from %s: "
1017                                "dststamp "LPX64" (current "LPX64")\n",
1018                                libcfs_id2str(lpid),
1019                                msg->ptlm_dststamp,
1020                                peer->peer_myincarnation);
1021                         kptllnd_peer_decref(peer);
1022                         return NULL;
1023                 }
1024
1025                 /* Brand new connection attempt: remove old incarnation */
1026                 kptllnd_peer_close_locked(peer, 0);
1027         }
1028
1029         kptllnd_cull_peertable_locked(lpid);
1030
1031         write_unlock_irqrestore(g_lock, flags);
1032
1033         if (peer != NULL) {
1034                 CDEBUG(D_NET, "Peer %s (%s) reconnecting:"
1035                        " stamp "LPX64"("LPX64")\n",
1036                        libcfs_id2str(lpid), kptllnd_ptlid2str(initiator),
1037                        msg->ptlm_srcstamp, peer->peer_incarnation);
1038
1039                 kptllnd_peer_decref(peer);
1040         }
1041
1042         hello_tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
1043         if (hello_tx == NULL) {
1044                 CERROR("Unable to allocate HELLO message for %s\n",
1045                        libcfs_id2str(lpid));
1046                 return NULL;
1047         }
1048
1049         kptllnd_init_msg(hello_tx->tx_msg, PTLLND_MSG_TYPE_HELLO,
1050                          sizeof(kptl_hello_msg_t));
1051
1052         new_peer = kptllnd_peer_allocate(lpid, initiator);
1053         if (new_peer == NULL) {
1054                 kptllnd_tx_decref(hello_tx);
1055                 return NULL;
1056         }
1057
1058         rc = kptllnd_peer_reserve_buffers();
1059         if (rc != 0) {
1060                 kptllnd_peer_decref(new_peer);
1061                 kptllnd_tx_decref(hello_tx);
1062
1063                 CERROR("Failed to reserve buffers for %s\n",
1064                        libcfs_id2str(lpid));
1065                 return NULL;
1066         }
1067
1068         write_lock_irqsave(g_lock, flags);
1069
1070  again:
1071         if (kptllnd_data.kptl_shutdown) {
1072                 write_unlock_irqrestore(g_lock, flags);
1073
1074                 CERROR ("Shutdown started, refusing connection from %s\n",
1075                         libcfs_id2str(lpid));
1076                 kptllnd_peer_unreserve_buffers();
1077                 kptllnd_peer_decref(new_peer);
1078                 kptllnd_tx_decref(hello_tx);
1079                 return NULL;
1080         }
1081
1082         peer = kptllnd_id2peer_locked(lpid);
1083         if (peer != NULL) {
1084                 if (peer->peer_state == PEER_STATE_WAITING_HELLO) {
1085                         /* An outgoing message instantiated 'peer' for me */
1086                         LASSERT(peer->peer_incarnation == 0);
1087
1088                         peer->peer_state = PEER_STATE_ACTIVE;
1089                         peer->peer_incarnation = msg->ptlm_srcstamp;
1090                         peer->peer_next_matchbits = safe_matchbits;
1091                         peer->peer_max_msg_size =
1092                                 msg->ptlm_u.hello.kptlhm_max_msg_size;
1093
1094                         write_unlock_irqrestore(g_lock, flags);
1095
1096                         CWARN("Outgoing instantiated peer %s\n",
1097                               libcfs_id2str(lpid));
1098                 } else {
1099                         LASSERT (peer->peer_state == PEER_STATE_ACTIVE);
1100
1101                         write_unlock_irqrestore(g_lock, flags);
1102
1103                         /* WOW!  Somehow this peer completed the HELLO
1104                          * handshake while I slept.  I guess I could have slept
1105                          * while it rebooted and sent a new HELLO, so I'll fail
1106                          * this one... */
1107                         CWARN("Wow! peer %s\n", libcfs_id2str(lpid));
1108                         kptllnd_peer_decref(peer);
1109                         peer = NULL;
1110                 }
1111
1112                 kptllnd_peer_unreserve_buffers();
1113                 kptllnd_peer_decref(new_peer);
1114                 kptllnd_tx_decref(hello_tx);
1115                 return peer;
1116         }
1117
1118         if (kptllnd_data.kptl_n_active_peers ==
1119             kptllnd_data.kptl_expected_peers) {
1120                 /* peer table full */
1121                 write_unlock_irqrestore(g_lock, flags);
1122
1123                 kptllnd_peertable_overflow_msg("Connection from ", lpid);
1124
1125                 rc = kptllnd_reserve_buffers(1); /* HELLO headroom */
1126                 if (rc != 0) {
1127                         CERROR("Refusing connection from %s\n",
1128                                libcfs_id2str(lpid));
1129                         kptllnd_peer_unreserve_buffers();
1130                         kptllnd_peer_decref(new_peer);
1131                         kptllnd_tx_decref(hello_tx);
1132                         return NULL;
1133                 }
1134                 
1135                 write_lock_irqsave(g_lock, flags);
1136                 kptllnd_data.kptl_expected_peers++;
1137                 goto again;
1138         }
1139
1140         last_matchbits_seen = kptllnd_get_last_seen_matchbits_locked(lpid);
1141
1142         hello_tx->tx_msg->ptlm_u.hello.kptlhm_matchbits = last_matchbits_seen;
1143         hello_tx->tx_msg->ptlm_u.hello.kptlhm_max_msg_size =
1144                 *kptllnd_tunables.kptl_max_msg_size;
1145
1146         new_peer->peer_state = PEER_STATE_ACTIVE;
1147         new_peer->peer_incarnation = msg->ptlm_srcstamp;
1148         new_peer->peer_next_matchbits = safe_matchbits;
1149         new_peer->peer_last_matchbits_seen = last_matchbits_seen;
1150         new_peer->peer_max_msg_size = msg->ptlm_u.hello.kptlhm_max_msg_size;
1151
1152         kptllnd_peer_add_peertable_locked(new_peer);
1153
1154         write_unlock_irqrestore(g_lock, flags);
1155
1156         /* NB someone else could get in now and post a message before I post
1157          * the HELLO, but post_tx/check_sends take care of that! */
1158
1159         CDEBUG(D_NETTRACE, "%s: post response hello %p\n",
1160                libcfs_id2str(new_peer->peer_id), hello_tx);
1161
1162         kptllnd_post_tx(new_peer, hello_tx, 0);
1163         kptllnd_peer_check_sends(new_peer);
1164
1165         return new_peer;
1166 }
1167
1168 void
1169 kptllnd_tx_launch(kptl_peer_t *peer, kptl_tx_t *tx, int nfrag)
1170 {
1171         kptllnd_post_tx(peer, tx, nfrag);
1172         kptllnd_peer_check_sends(peer);
1173 }
1174
1175 int
1176 kptllnd_find_target(kptl_peer_t **peerp, lnet_process_id_t target)
1177 {
1178         rwlock_t         *g_lock = &kptllnd_data.kptl_peer_rw_lock;
1179         ptl_process_id_t  ptl_id;
1180         kptl_peer_t      *new_peer;
1181         kptl_tx_t        *hello_tx;
1182         unsigned long     flags;
1183         int               rc;
1184         __u64             last_matchbits_seen;
1185
1186         /* I expect to find the peer, so I only take a read lock... */
1187         read_lock_irqsave(g_lock, flags);
1188         *peerp = kptllnd_id2peer_locked(target);
1189         read_unlock_irqrestore(g_lock, flags);
1190
1191         if (*peerp != NULL)
1192                 return 0;
1193         
1194         if ((target.pid & LNET_PID_USERFLAG) != 0) {
1195                 CWARN("Refusing to create a new connection to %s "
1196                       "(non-kernel peer)\n", libcfs_id2str(target));
1197                 return -EHOSTUNREACH;
1198         }
1199
1200         /* The new peer is a kernel ptllnd, and kernel ptllnds all have
1201          * the same portals PID */
1202         ptl_id.nid = kptllnd_lnet2ptlnid(target.nid);
1203         ptl_id.pid = kptllnd_data.kptl_portals_id.pid;
1204
1205         hello_tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
1206         if (hello_tx == NULL) {
1207                 CERROR("Unable to allocate connect message for %s\n",
1208                        libcfs_id2str(target));
1209                 return -ENOMEM;
1210         }
1211
1212         kptllnd_init_msg(hello_tx->tx_msg, PTLLND_MSG_TYPE_HELLO,
1213                          sizeof(kptl_hello_msg_t));
1214
1215         new_peer = kptllnd_peer_allocate(target, ptl_id);
1216         if (new_peer == NULL) {
1217                 rc = -ENOMEM;
1218                 goto unwind_0;
1219         }
1220
1221         rc = kptllnd_peer_reserve_buffers();
1222         if (rc != 0)
1223                 goto unwind_1;
1224
1225         write_lock_irqsave(g_lock, flags);
1226  again:
1227         if (kptllnd_data.kptl_shutdown) {
1228                 write_unlock_irqrestore(g_lock, flags);
1229                 rc = -ESHUTDOWN;
1230                 goto unwind_2;
1231         }
1232
1233         *peerp = kptllnd_id2peer_locked(target);
1234         if (*peerp != NULL) {
1235                 write_unlock_irqrestore(g_lock, flags);
1236                 goto unwind_2;
1237         }
1238
1239         kptllnd_cull_peertable_locked(target);
1240
1241         if (kptllnd_data.kptl_n_active_peers ==
1242             kptllnd_data.kptl_expected_peers) {
1243                 /* peer table full */
1244                 write_unlock_irqrestore(g_lock, flags);
1245
1246                 kptllnd_peertable_overflow_msg("Connection to ", target);
1247
1248                 rc = kptllnd_reserve_buffers(1); /* HELLO headroom */
1249                 if (rc != 0) {
1250                         CERROR("Can't create connection to %s\n",
1251                                libcfs_id2str(target));
1252                         rc = -ENOMEM;
1253                         goto unwind_2;
1254                 }
1255                 write_lock_irqsave(g_lock, flags);
1256                 kptllnd_data.kptl_expected_peers++;
1257                 goto again;
1258         }
1259
1260         last_matchbits_seen = kptllnd_get_last_seen_matchbits_locked(target);
1261
1262         hello_tx->tx_msg->ptlm_u.hello.kptlhm_matchbits = last_matchbits_seen;
1263         hello_tx->tx_msg->ptlm_u.hello.kptlhm_max_msg_size =
1264                 *kptllnd_tunables.kptl_max_msg_size;
1265                 
1266         new_peer->peer_state = PEER_STATE_WAITING_HELLO;
1267         new_peer->peer_last_matchbits_seen = last_matchbits_seen;
1268         
1269         kptllnd_peer_add_peertable_locked(new_peer);
1270
1271         write_unlock_irqrestore(g_lock, flags);
1272
1273         /* NB someone else could get in now and post a message before I post
1274          * the HELLO, but post_tx/check_sends take care of that! */
1275
1276         CDEBUG(D_NETTRACE, "%s: post initial hello %p\n",
1277                libcfs_id2str(new_peer->peer_id), hello_tx);
1278
1279         kptllnd_post_tx(new_peer, hello_tx, 0);
1280         kptllnd_peer_check_sends(new_peer);
1281        
1282         *peerp = new_peer;
1283         return 0;
1284         
1285  unwind_2:
1286         kptllnd_peer_unreserve_buffers();
1287  unwind_1:
1288         kptllnd_peer_decref(new_peer);
1289  unwind_0:
1290         kptllnd_tx_decref(hello_tx);
1291
1292         return rc;
1293 }