Whamcloud - gitweb
* Applied the last patch in Bug 2306, which changes the portals router/NAL
[fs/lustre-release.git] / lnet / klnds / socklnd / socklnd_cb.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
5  *   Author: Zach Brown <zab@zabbo.net>
6  *   Author: Peter J. Braam <braam@clusterfs.com>
7  *   Author: Phil Schwan <phil@clusterfs.com>
8  *   Author: Eric Barton <eric@bartonsoftware.com>
9  *
10  *   This file is part of Portals, http://www.sf.net/projects/sandiaportals/
11  *
12  *   Portals is free software; you can redistribute it and/or
13  *   modify it under the terms of version 2 of the GNU General Public
14  *   License as published by the Free Software Foundation.
15  *
16  *   Portals is distributed in the hope that it will be useful,
17  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
18  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
19  *   GNU General Public License for more details.
20  *
21  *   You should have received a copy of the GNU General Public License
22  *   along with Portals; if not, write to the Free Software
23  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24  */
25
26 #include "socknal.h"
27
28 /*
29  *  LIB functions follow
30  *
31  */
32 ptl_err_t
33 ksocknal_read(nal_cb_t *nal, void *private, void *dst_addr,
34               user_ptr src_addr, size_t len)
35 {
36         CDEBUG(D_NET, LPX64": reading %ld bytes from %p -> %p\n",
37                nal->ni.nid, (long)len, src_addr, dst_addr);
38
39         memcpy( dst_addr, src_addr, len );
40         return PTL_OK;
41 }
42
43 ptl_err_t
44 ksocknal_write(nal_cb_t *nal, void *private, user_ptr dst_addr,
45                void *src_addr, size_t len)
46 {
47         CDEBUG(D_NET, LPX64": writing %ld bytes from %p -> %p\n",
48                nal->ni.nid, (long)len, src_addr, dst_addr);
49
50         memcpy( dst_addr, src_addr, len );
51         return PTL_OK;
52 }
53
54 void *
55 ksocknal_malloc(nal_cb_t *nal, size_t len)
56 {
57         void *buf;
58
59         PORTAL_ALLOC(buf, len);
60
61         if (buf != NULL)
62                 memset(buf, 0, len);
63
64         return (buf);
65 }
66
67 void
68 ksocknal_free(nal_cb_t *nal, void *buf, size_t len)
69 {
70         PORTAL_FREE(buf, len);
71 }
72
73 void
74 ksocknal_printf(nal_cb_t *nal, const char *fmt, ...)
75 {
76         va_list ap;
77         char msg[256];
78
79         va_start (ap, fmt);
80         vsnprintf (msg, sizeof (msg), fmt, ap); /* sprint safely */
81         va_end (ap);
82
83         msg[sizeof (msg) - 1] = 0;              /* ensure terminated */
84
85         CDEBUG (D_NET, "%s", msg);
86 }
87
88 void
89 ksocknal_cli(nal_cb_t *nal, unsigned long *flags)
90 {
91         ksock_nal_data_t *data = nal->nal_data;
92
93         spin_lock(&data->ksnd_nal_cb_lock);
94 }
95
96 void
97 ksocknal_sti(nal_cb_t *nal, unsigned long *flags)
98 {
99         ksock_nal_data_t *data;
100         data = nal->nal_data;
101
102         spin_unlock(&data->ksnd_nal_cb_lock);
103 }
104
105 int
106 ksocknal_dist(nal_cb_t *nal, ptl_nid_t nid, unsigned long *dist)
107 {
108         /* I would guess that if ksocknal_get_peer (nid) == NULL,
109            and we're not routing, then 'nid' is very distant :) */
110         if ( nal->ni.nid == nid ) {
111                 *dist = 0;
112         } else {
113                 *dist = 1;
114         }
115
116         return 0;
117 }
118
119 void
120 ksocknal_free_ltx (ksock_ltx_t *ltx)
121 {
122         atomic_dec(&ksocknal_data.ksnd_nactive_ltxs);
123         PORTAL_FREE(ltx, ltx->ltx_desc_size);
124 }
125
126 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
127 struct page *
128 ksocknal_kvaddr_to_page (unsigned long vaddr)
129 {
130         struct page *page;
131
132         if (vaddr >= VMALLOC_START &&
133             vaddr < VMALLOC_END)
134                 page = vmalloc_to_page ((void *)vaddr);
135 #if CONFIG_HIGHMEM
136         else if (vaddr >= PKMAP_BASE &&
137                  vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
138                 page = vmalloc_to_page ((void *)vaddr);
139                 /* in 2.4 ^ just walks the page tables */
140 #endif
141         else
142                 page = virt_to_page (vaddr);
143
144         if (page == NULL ||
145             !VALID_PAGE (page))
146                 return (NULL);
147
148         return (page);
149 }
150 #endif
151
152 int
153 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
154 {
155         struct socket *sock = conn->ksnc_sock;
156         struct iovec  *iov = tx->tx_iov;
157         int            fragsize = iov->iov_len;
158         unsigned long  vaddr = (unsigned long)iov->iov_base;
159         int            more = (tx->tx_niov > 1) || 
160                               (tx->tx_nkiov > 0) ||
161                               (!list_empty (&conn->ksnc_tx_queue));
162 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
163         int            offset = vaddr & (PAGE_SIZE - 1);
164         int            zcsize = MIN (fragsize, PAGE_SIZE - offset);
165         struct page   *page;
166 #endif
167         int            rc;
168
169         /* NB we can't trust socket ops to either consume our iovs
170          * or leave them alone, so we only send 1 frag at a time. */
171         LASSERT (fragsize <= tx->tx_resid);
172         LASSERT (tx->tx_niov > 0);
173         
174 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
175         if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
176             (sock->sk->route_caps & NETIF_F_SG) &&
177             (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
178             (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
179                 
180                 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
181                        (void *)vaddr, page, page_address(page), offset, zcsize);
182
183                 if (fragsize > zcsize) {
184                         more = 1;
185                         fragsize = zcsize;
186                 }
187
188                 rc = tcp_sendpage_zccd(sock, page, offset, zcsize, 
189                                        more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
190                                        &tx->tx_zccd);
191         } else
192 #endif
193         {
194                 /* NB don't pass tx's iov; sendmsg may or may not update it */
195                 struct iovec fragiov = { .iov_base = (void *)vaddr,
196                                          .iov_len  = fragsize};
197                 struct msghdr msg = {
198                         .msg_name       = NULL,
199                         .msg_namelen    = 0,
200                         .msg_iov        = &fragiov,
201                         .msg_iovlen     = 1,
202                         .msg_control    = NULL,
203                         .msg_controllen = 0,
204                         .msg_flags      = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
205                 };
206                 mm_segment_t oldmm = get_fs();
207
208                 set_fs (KERNEL_DS);
209                 rc = sock_sendmsg(sock, &msg, fragsize);
210                 set_fs (oldmm);
211         } 
212
213         if (rc > 0) {
214                 tx->tx_resid -= rc;
215
216                 if (rc < iov->iov_len) {
217                         /* didn't send whole iov entry... */
218                         iov->iov_base = (void *)(vaddr + rc);
219                         iov->iov_len -= rc;
220                 } else {
221                         tx->tx_iov++;
222                         tx->tx_niov--;
223                 }
224         }
225         
226         return (rc);
227 }
228
229 int
230 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
231 {
232         struct socket *sock = conn->ksnc_sock;
233         ptl_kiov_t    *kiov = tx->tx_kiov;
234         int            fragsize = kiov->kiov_len;
235         struct page   *page = kiov->kiov_page;
236         int            offset = kiov->kiov_offset;
237         int            more = (tx->tx_nkiov > 1) ||
238                               (!list_empty (&conn->ksnc_tx_queue));
239         int            rc;
240
241         /* NB we can't trust socket ops to either consume our iovs
242          * or leave them alone, so we only send 1 frag at a time. */
243         LASSERT (fragsize <= tx->tx_resid);
244         LASSERT (offset + fragsize <= PAGE_SIZE);
245         LASSERT (tx->tx_niov == 0);
246         LASSERT (tx->tx_nkiov > 0);
247
248 #if SOCKNAL_ZC
249         if (fragsize >= ksocknal_data.ksnd_zc_min_frag &&
250             (sock->sk->route_caps & NETIF_F_SG) &&
251             (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
252
253                 CDEBUG(D_NET, "page %p + offset %x for %d\n",
254                                page, offset, fragsize);
255
256                 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
257                                        more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
258                                        &tx->tx_zccd);
259         } else
260 #endif
261         {
262                 char *addr = ((char *)kmap (page)) + offset;
263                 struct iovec fragiov = {.iov_base = addr,
264                                         .iov_len  = fragsize};
265                 struct msghdr msg = {
266                         .msg_name       = NULL,
267                         .msg_namelen    = 0,
268                         .msg_iov        = &fragiov,
269                         .msg_iovlen     = 1,
270                         .msg_control    = NULL,
271                         .msg_controllen = 0,
272                         .msg_flags      = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
273                 };
274                 mm_segment_t  oldmm = get_fs();
275                 
276                 set_fs (KERNEL_DS);
277                 rc = sock_sendmsg(sock, &msg, fragsize);
278                 set_fs (oldmm);
279
280                 kunmap (page);
281         }
282
283         if (rc > 0) {
284                 tx->tx_resid -= rc;
285  
286                 if (rc < fragsize) {
287                         kiov->kiov_offset = offset + rc;
288                         kiov->kiov_len    = fragsize - rc;
289                 } else {
290                         tx->tx_kiov++;
291                         tx->tx_nkiov--;
292                 }
293         }
294
295         return (rc);
296 }
297
298 int
299 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
300 {
301         int      rc;
302         
303         if (ksocknal_data.ksnd_stall_tx != 0) {
304                 set_current_state (TASK_UNINTERRUPTIBLE);
305                 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
306         }
307
308         LASSERT (tx->tx_resid != 0);
309
310         rc = ksocknal_getconnsock (conn);
311         if (rc != 0) {
312                 LASSERT (conn->ksnc_closing);
313                 return (-ESHUTDOWN);
314         }
315
316         do {
317                 if (ksocknal_data.ksnd_enomem_tx > 0) {
318                         /* testing... */
319                         ksocknal_data.ksnd_enomem_tx--;
320                         rc = -EAGAIN;
321                 } else if (tx->tx_niov != 0) {
322                         rc = ksocknal_send_iov (conn, tx);
323                 } else {
324                         rc = ksocknal_send_kiov (conn, tx);
325                 }
326
327                 if (rc <= 0) {
328                         /* Didn't write anything.
329                          *
330                          * NB: rc == 0 and rc == -EAGAIN both mean try
331                          * again later (linux stack returns -EAGAIN for
332                          * this, but Adaptech TOE returns 0).
333                          *
334                          * Also, sends never fail with -ENOMEM, just
335                          * -EAGAIN, but with the added bonus that we can't
336                          * expect write_space() to call us back to tell us
337                          * when to try sending again.  We use the
338                          * SOCK_NOSPACE flag to diagnose...  */
339
340                         LASSERT(rc != -ENOMEM);
341
342                         if (rc == 0 || rc == -EAGAIN) {
343                                 if (test_bit(SOCK_NOSPACE, 
344                                              &conn->ksnc_sock->flags)) {
345                                         rc = -EAGAIN;
346                                 } else {
347                                         static int counter;
348                          
349                                         counter++;
350                                         if ((counter & (-counter)) == counter)
351                                                 CWARN("%d ENOMEM tx %p\n", 
352                                                       counter, conn);
353                                         rc = -ENOMEM;
354                                 }
355                         }
356                         break;
357                 }
358
359                 rc = 0;
360
361                 /* Consider the connection alive since we managed to chuck
362                  * more data into it.  Really, we'd like to consider it
363                  * alive only when the peer ACKs something, but
364                  * write_space() only gets called back while SOCK_NOSPACE
365                  * is set.  Instead, we presume peer death has occurred if
366                  * the socket doesn't drain within a timout */
367                 conn->ksnc_tx_deadline = jiffies + 
368                                          ksocknal_data.ksnd_io_timeout * HZ;
369                 conn->ksnc_peer->ksnp_last_alive = jiffies;
370
371         } while (tx->tx_resid != 0);
372
373         ksocknal_putconnsock (conn);
374         return (rc);
375 }
376
377 void
378 ksocknal_eager_ack (ksock_conn_t *conn)
379 {
380         int            opt = 1;
381         mm_segment_t   oldmm = get_fs();
382         struct socket *sock = conn->ksnc_sock;
383         
384         /* Remind the socket to ACK eagerly.  If I don't, the socket might
385          * think I'm about to send something it could piggy-back the ACK
386          * on, introducing delay in completing zero-copy sends in my
387          * peer. */
388
389         set_fs(KERNEL_DS);
390         sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
391                                (char *)&opt, sizeof (opt));
392         set_fs(oldmm);
393 }
394
395 int
396 ksocknal_recv_iov (ksock_conn_t *conn)
397 {
398         struct iovec *iov = conn->ksnc_rx_iov;
399         int           fragsize  = iov->iov_len;
400         unsigned long vaddr = (unsigned long)iov->iov_base;
401         struct iovec  fragiov = { .iov_base = (void *)vaddr,
402                                   .iov_len  = fragsize};
403         struct msghdr msg = {
404                 .msg_name       = NULL,
405                 .msg_namelen    = 0,
406                 .msg_iov        = &fragiov,
407                 .msg_iovlen     = 1,
408                 .msg_control    = NULL,
409                 .msg_controllen = 0,
410                 .msg_flags      = 0
411         };
412         mm_segment_t oldmm = get_fs();
413         int          rc;
414
415         /* NB we can't trust socket ops to either consume our iovs
416          * or leave them alone, so we only receive 1 frag at a time. */
417         LASSERT (conn->ksnc_rx_niov > 0);
418         LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
419
420         set_fs (KERNEL_DS);
421         rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
422         /* NB this is just a boolean............................^ */
423         set_fs (oldmm);
424
425         if (rc <= 0)
426                 return (rc);
427
428         /* received something... */
429         conn->ksnc_peer->ksnp_last_alive = jiffies;
430         conn->ksnc_rx_deadline = jiffies + 
431                                  ksocknal_data.ksnd_io_timeout * HZ;
432         mb();                           /* order with setting rx_started */
433         conn->ksnc_rx_started = 1;
434
435         conn->ksnc_rx_nob_wanted -= rc;
436         conn->ksnc_rx_nob_left -= rc;
437                 
438         if (rc < fragsize) {
439                 iov->iov_base = (void *)(vaddr + rc);
440                 iov->iov_len = fragsize - rc;
441                 return (-EAGAIN);
442         }
443
444         conn->ksnc_rx_iov++;
445         conn->ksnc_rx_niov--;
446         return (1);
447 }
448
449 int
450 ksocknal_recv_kiov (ksock_conn_t *conn)
451 {
452         ptl_kiov_t   *kiov = conn->ksnc_rx_kiov;
453         struct page  *page = kiov->kiov_page;
454         int           offset = kiov->kiov_offset;
455         int           fragsize = kiov->kiov_len;
456         unsigned long vaddr = ((unsigned long)kmap (page)) + offset;
457         struct iovec  fragiov = { .iov_base = (void *)vaddr,
458                                   .iov_len  = fragsize};
459         struct msghdr msg = {
460                 .msg_name       = NULL,
461                 .msg_namelen    = 0,
462                 .msg_iov        = &fragiov,
463                 .msg_iovlen     = 1,
464                 .msg_control    = NULL,
465                 .msg_controllen = 0,
466                 .msg_flags      = 0
467         };
468         mm_segment_t oldmm = get_fs();
469         int          rc;
470
471         /* NB we can't trust socket ops to either consume our iovs
472          * or leave them alone, so we only receive 1 frag at a time. */
473         LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
474         LASSERT (conn->ksnc_rx_nkiov > 0);
475         LASSERT (offset + fragsize <= PAGE_SIZE);
476
477         set_fs (KERNEL_DS);
478         rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
479         /* NB this is just a boolean............................^ */
480         set_fs (oldmm);
481
482         kunmap (page);
483         
484         if (rc <= 0)
485                 return (rc);
486         
487         /* received something... */
488         conn->ksnc_peer->ksnp_last_alive = jiffies;
489         conn->ksnc_rx_deadline = jiffies + 
490                                  ksocknal_data.ksnd_io_timeout * HZ;
491         mb();                           /* order with setting rx_started */
492         conn->ksnc_rx_started = 1;
493
494         conn->ksnc_rx_nob_wanted -= rc;
495         conn->ksnc_rx_nob_left -= rc;
496                 
497         if (rc < fragsize) {
498                 kiov->kiov_offset = offset + rc;
499                 kiov->kiov_len = fragsize - rc;
500                 return (-EAGAIN);
501         }
502
503         conn->ksnc_rx_kiov++;
504         conn->ksnc_rx_nkiov--;
505         return (1);
506 }
507
508 int
509 ksocknal_receive (ksock_conn_t *conn) 
510 {
511         /* Return 1 on success, 0 on EOF, < 0 on error.
512          * Caller checks ksnc_rx_nob_wanted to determine
513          * progress/completion. */
514         int     rc;
515         ENTRY;
516         
517         if (ksocknal_data.ksnd_stall_rx != 0) {
518                 set_current_state (TASK_UNINTERRUPTIBLE);
519                 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
520         }
521
522         rc = ksocknal_getconnsock (conn);
523         if (rc != 0) {
524                 LASSERT (conn->ksnc_closing);
525                 return (-ESHUTDOWN);
526         }
527
528         for (;;) {
529                 if (conn->ksnc_rx_niov != 0)
530                         rc = ksocknal_recv_iov (conn);
531                 else
532                         rc = ksocknal_recv_kiov (conn);
533
534                 if (rc <= 0) {
535                         /* error/EOF or partial receive */
536                         if (rc == -EAGAIN) {
537                                 rc = 1;
538                         } else if (rc == 0 && conn->ksnc_rx_started) {
539                                 /* EOF in the middle of a message */
540                                 rc = -EPROTO;
541                         }
542                         break;
543                 }
544
545                 /* Completed a fragment */
546
547                 if (conn->ksnc_rx_nob_wanted == 0) {
548                         /* Completed a message segment (header or payload) */
549                         if ((ksocknal_data.ksnd_eager_ack & conn->ksnc_type) != 0 &&
550                             (conn->ksnc_rx_state ==  SOCKNAL_RX_BODY ||
551                              conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
552                                 /* Remind the socket to ack eagerly... */
553                                 ksocknal_eager_ack(conn);
554                         }
555                         rc = 1;
556                         break;
557                 }
558         }
559
560         ksocknal_putconnsock (conn);
561         RETURN (rc);
562 }
563
564 #if SOCKNAL_ZC
565 void
566 ksocknal_zc_callback (zccd_t *zcd)
567 {
568         ksock_tx_t    *tx = KSOCK_ZCCD_2_TX(zcd);
569         ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
570         unsigned long  flags;
571         ENTRY;
572
573         /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
574
575         spin_lock_irqsave (&sched->kss_lock, flags);
576
577         list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
578         wake_up (&sched->kss_waitq);
579
580         spin_unlock_irqrestore (&sched->kss_lock, flags);
581         EXIT;
582 }
583 #endif
584
585 void
586 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
587 {
588         ksock_ltx_t   *ltx;
589         ENTRY;
590
591         if (tx->tx_conn != NULL) {
592                 /* This tx got queued on a conn; do the accounting... */
593                 atomic_sub (tx->tx_nob, &tx->tx_conn->ksnc_tx_nob);
594 #if SOCKNAL_ZC
595                 /* zero copy completion isn't always from
596                  * process_transmit() so it needs to keep a ref on
597                  * tx_conn... */
598                 if (asynch)
599                         ksocknal_put_conn (tx->tx_conn);
600 #else
601                 LASSERT (!asynch);
602 #endif
603         }
604
605         if (tx->tx_isfwd) {             /* was a forwarded packet? */
606                 kpr_fwd_done (&ksocknal_data.ksnd_router,
607                               KSOCK_TX_2_KPR_FWD_DESC (tx), 
608                               (tx->tx_resid == 0) ? 0 : -ECONNABORTED);
609                 EXIT;
610                 return;
611         }
612
613         /* local send */
614         ltx = KSOCK_TX_2_KSOCK_LTX (tx);
615
616         lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie,
617                       (tx->tx_resid == 0) ? PTL_OK : PTL_FAIL);
618
619         ksocknal_free_ltx (ltx);
620         EXIT;
621 }
622
623 void
624 ksocknal_tx_launched (ksock_tx_t *tx) 
625 {
626 #if SOCKNAL_ZC
627         if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
628                 ksock_conn_t  *conn = tx->tx_conn;
629                 
630                 /* zccd skbufs are still in-flight.  First take a ref on
631                  * conn, so it hangs about for ksocknal_tx_done... */
632                 atomic_inc (&conn->ksnc_refcount);
633
634                 /* ...then drop the initial ref on zccd, so the zero copy
635                  * callback can occur */
636                 zccd_put (&tx->tx_zccd);
637                 return;
638         }
639 #endif
640         /* Any zero-copy-ness (if any) has completed; I can complete the
641          * transmit now, avoiding an extra schedule */
642         ksocknal_tx_done (tx, 0);
643 }
644
645 int
646 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
647 {
648         unsigned long  flags;
649         int            rc;
650        
651         rc = ksocknal_transmit (conn, tx);
652
653         CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
654
655         if (tx->tx_resid == 0) {
656                 /* Sent everything OK */
657                 LASSERT (rc == 0);
658
659                 ksocknal_tx_launched (tx);
660                 return (0);
661         }
662
663         if (rc == -EAGAIN)
664                 return (rc);
665
666         if (rc == -ENOMEM) {
667                 /* Queue on ksnd_enomem_conns for retry after a timeout */
668                 spin_lock_irqsave(&ksocknal_data.ksnd_reaper_lock, flags);
669
670                 /* enomem list takes over scheduler's ref... */
671                 LASSERT (conn->ksnc_tx_scheduled);
672                 list_add_tail(&conn->ksnc_tx_list,
673                               &ksocknal_data.ksnd_enomem_conns);
674                 if (!time_after_eq(jiffies + SOCKNAL_ENOMEM_RETRY,
675                                    ksocknal_data.ksnd_reaper_waketime))
676                         wake_up (&ksocknal_data.ksnd_reaper_waitq);
677                 
678                 spin_unlock_irqrestore(&ksocknal_data.ksnd_reaper_lock, flags);
679                 return (rc);
680         }
681
682         /* Actual error */
683         LASSERT (rc < 0);
684
685         if (!conn->ksnc_closing)
686                 CERROR("[%p] Error %d on write to "LPX64
687                        " ip %d.%d.%d.%d:%d\n", conn, rc,
688                        conn->ksnc_peer->ksnp_nid,
689                        HIPQUAD(conn->ksnc_ipaddr),
690                        conn->ksnc_port);
691
692         ksocknal_close_conn_and_siblings (conn, rc);
693         ksocknal_tx_launched (tx);
694
695         return (rc);
696 }
697
698 void
699 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
700 {
701         unsigned long     flags;
702
703         /* called holding write lock on ksnd_global_lock */
704
705         LASSERT (!route->ksnr_deleted);
706         LASSERT ((route->ksnr_connected & (1 << SOCKNAL_CONN_ANY)) == 0);
707         LASSERT ((route->ksnr_connected & KSNR_TYPED_ROUTES) != KSNR_TYPED_ROUTES);
708         LASSERT (!route->ksnr_connecting);
709         
710         if (ksocknal_data.ksnd_typed_conns)
711                 route->ksnr_connecting = 
712                         KSNR_TYPED_ROUTES & ~route->ksnr_connected;
713         else
714                 route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
715
716         atomic_inc (&route->ksnr_refcount);     /* extra ref for asynchd */
717         
718         spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
719         
720         list_add_tail (&route->ksnr_connect_list,
721                        &ksocknal_data.ksnd_autoconnectd_routes);
722         wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
723         
724         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
725 }
726
727 ksock_peer_t *
728 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
729 {
730         char          ipbuf[PTL_NALFMT_SIZE];
731         ptl_nid_t     target_nid;
732         int           rc;
733         ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
734
735         if (peer != NULL)
736                 return (peer);
737
738         if (tx->tx_isfwd) {
739                 CERROR ("Can't send packet to "LPX64
740                        " %s: routed target is not a peer\n",
741                         nid, portals_nid2str(SOCKNAL, nid, ipbuf));
742                 return (NULL);
743         }
744
745         rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
746                          &target_nid);
747         if (rc != 0) {
748                 CERROR ("Can't route to "LPX64" %s: router error %d\n",
749                         nid, portals_nid2str(SOCKNAL, nid, ipbuf), rc);
750                 return (NULL);
751         }
752
753         peer = ksocknal_find_peer_locked (target_nid);
754         if (peer != NULL)
755                 return (peer);
756
757         CERROR ("Can't send packet to "LPX64" %s: no peer entry\n",
758                 target_nid, portals_nid2str(SOCKNAL, target_nid, ipbuf));
759         return (NULL);
760 }
761
762 ksock_conn_t *
763 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer) 
764 {
765         struct list_head *tmp;
766         ksock_conn_t     *typed = NULL;
767         int               tnob  = 0;
768         ksock_conn_t     *fallback = NULL;
769         int               fnob     = 0;
770         
771         /* Find the conn with the shortest tx queue */
772         list_for_each (tmp, &peer->ksnp_conns) {
773                 ksock_conn_t *c = list_entry(tmp, ksock_conn_t, ksnc_list);
774                 int           nob = atomic_read(&c->ksnc_tx_nob) +
775                                     c->ksnc_sock->sk->sk_wmem_queued;
776
777                 LASSERT (!c->ksnc_closing);
778
779                 if (fallback == NULL || nob < fnob) {
780                         fallback = c;
781                         fnob     = nob;
782                 }
783
784                 if (!ksocknal_data.ksnd_typed_conns)
785                         continue;
786
787                 switch (c->ksnc_type) {
788                 default:
789                         LBUG();
790                 case SOCKNAL_CONN_ANY:
791                         break;
792                 case SOCKNAL_CONN_BULK_IN:
793                         continue;
794                 case SOCKNAL_CONN_BULK_OUT:
795                         if (tx->tx_nob < ksocknal_data.ksnd_min_bulk)
796                                 continue;
797                         break;
798                 case SOCKNAL_CONN_CONTROL:
799                         if (tx->tx_nob >= ksocknal_data.ksnd_min_bulk)
800                                 continue;
801                         break;
802                 }
803
804                 if (typed == NULL || nob < tnob) {
805                         typed = c;
806                         tnob  = nob;
807                 }
808         }
809
810         /* prefer the typed selection */
811         return ((typed != NULL) ? typed : fallback);
812 }
813
814 void
815 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
816 {
817         unsigned long  flags;
818         ksock_sched_t *sched = conn->ksnc_scheduler;
819
820         /* called holding global lock (read or irq-write) and caller may
821          * not have dropped this lock between finding conn and calling me,
822          * so we don't need the {get,put}connsock dance to deref
823          * ksnc_sock... */
824         LASSERT(!conn->ksnc_closing);
825         LASSERT(tx->tx_resid == tx->tx_nob);
826         
827         CDEBUG (D_NET, "Sending to "LPX64" ip %d.%d.%d.%d:%d\n", 
828                 conn->ksnc_peer->ksnp_nid,
829                 HIPQUAD(conn->ksnc_ipaddr),
830                 conn->ksnc_port);
831
832         atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
833         tx->tx_conn = conn;
834
835 #if SOCKNAL_ZC
836         zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
837         /* NB this sets 1 ref on zccd, so the callback can only occur after
838          * I've released this ref. */
839 #endif
840         spin_lock_irqsave (&sched->kss_lock, flags);
841
842         conn->ksnc_tx_deadline = jiffies + 
843                                  ksocknal_data.ksnd_io_timeout * HZ;
844         mb();                                   /* order with list_add_tail */
845
846         list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
847                 
848         if (conn->ksnc_tx_ready &&      /* able to send */
849             !conn->ksnc_tx_scheduled) { /* not scheduled to send */
850                 /* +1 ref for scheduler */
851                 atomic_inc (&conn->ksnc_refcount);
852                 list_add_tail (&conn->ksnc_tx_list, 
853                                &sched->kss_tx_conns);
854                 conn->ksnc_tx_scheduled = 1;
855                 wake_up (&sched->kss_waitq);
856         }
857
858         spin_unlock_irqrestore (&sched->kss_lock, flags);
859 }
860
861 ksock_route_t *
862 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
863 {
864         struct list_head  *tmp;
865         ksock_route_t     *route;
866         ksock_route_t     *candidate = NULL;
867         int                found = 0;
868         int                bits;
869         
870         list_for_each (tmp, &peer->ksnp_routes) {
871                 route = list_entry (tmp, ksock_route_t, ksnr_list);
872                 bits  = route->ksnr_connected;
873                 
874                 if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES ||
875                     (bits & (1 << SOCKNAL_CONN_ANY)) != 0 ||
876                     route->ksnr_connecting != 0) {
877                         /* All typed connections have been established, or
878                          * an untyped connection has been established, or
879                          * connections are currently being established */
880                         found = 1;
881                         continue;
882                 }
883
884                 /* too soon to retry this guy? */
885                 if (!time_after_eq (jiffies, route->ksnr_timeout))
886                         continue;
887                 
888                 /* always do eager routes */
889                 if (route->ksnr_eager)
890                         return (route);
891
892                 if (candidate == NULL) {
893                         /* If we don't find any other route that is fully
894                          * connected or connecting, the first connectable
895                          * route is returned.  If it fails to connect, it
896                          * will get placed at the end of the list */
897                         candidate = route;
898                 }
899         }
900  
901         return (found ? NULL : candidate);
902 }
903
904 ksock_route_t *
905 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
906 {
907         struct list_head  *tmp;
908         ksock_route_t     *route;
909
910         list_for_each (tmp, &peer->ksnp_routes) {
911                 route = list_entry (tmp, ksock_route_t, ksnr_list);
912                 
913                 if (route->ksnr_connecting != 0)
914                         return (route);
915         }
916         
917         return (NULL);
918 }
919
920 int
921 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
922 {
923         unsigned long     flags;
924         ksock_peer_t     *peer;
925         ksock_conn_t     *conn;
926         ksock_route_t    *route;
927         rwlock_t         *g_lock;
928         
929         /* Ensure the frags we've been given EXACTLY match the number of
930          * bytes we want to send.  Many TCP/IP stacks disregard any total
931          * size parameters passed to them and just look at the frags. 
932          *
933          * We always expect at least 1 mapped fragment containing the
934          * complete portals header. */
935         LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
936                  lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
937         LASSERT (tx->tx_niov >= 1);
938         LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
939
940         CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
941                 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type, 
942                 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
943
944         tx->tx_conn = NULL;                     /* only set when assigned a conn */
945         tx->tx_resid = tx->tx_nob;
946         tx->tx_hdr = (ptl_hdr_t *)tx->tx_iov[0].iov_base;
947
948         g_lock = &ksocknal_data.ksnd_global_lock;
949         read_lock (g_lock);
950         
951         peer = ksocknal_find_target_peer_locked (tx, nid);
952         if (peer == NULL) {
953                 read_unlock (g_lock);
954                 return (-EHOSTUNREACH);
955         }
956
957         if (ksocknal_find_connectable_route_locked(peer) == NULL) {
958                 conn = ksocknal_find_conn_locked (tx, peer);
959                 if (conn != NULL) {
960                         /* I've got no autoconnect routes that need to be
961                          * connecting and I do have an actual connection... */
962                         ksocknal_queue_tx_locked (tx, conn);
963                         read_unlock (g_lock);
964                         return (0);
965                 }
966         }
967         
968         /* Making one or more connections; I'll need a write lock... */
969
970         atomic_inc (&peer->ksnp_refcount);      /* +1 ref for me while I unlock */
971         read_unlock (g_lock);
972         write_lock_irqsave (g_lock, flags);
973         
974         if (peer->ksnp_closing) {               /* peer deleted as I blocked! */
975                 write_unlock_irqrestore (g_lock, flags);
976                 ksocknal_put_peer (peer);
977                 return (-EHOSTUNREACH);
978         }
979         ksocknal_put_peer (peer);               /* drop ref I got above */
980
981         for (;;) {
982                 /* launch any/all autoconnections that need it */
983                 route = ksocknal_find_connectable_route_locked (peer);
984                 if (route == NULL)
985                         break;
986
987                 ksocknal_launch_autoconnect_locked (route);
988         }
989
990         conn = ksocknal_find_conn_locked (tx, peer);
991         if (conn != NULL) {
992                 /* Connection exists; queue message on it */
993                 ksocknal_queue_tx_locked (tx, conn);
994                 write_unlock_irqrestore (g_lock, flags);
995                 return (0);
996         }
997
998         route = ksocknal_find_connecting_route_locked (peer);
999         if (route != NULL) {
1000                 /* At least 1 connection is being established; queue the
1001                  * message... */
1002                 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
1003                 write_unlock_irqrestore (g_lock, flags);
1004                 return (0);
1005         }
1006         
1007         write_unlock_irqrestore (g_lock, flags);
1008         return (-EHOSTUNREACH);
1009 }
1010
1011 ptl_err_t
1012 ksocknal_sendmsg(nal_cb_t     *nal, 
1013                  void         *private, 
1014                  lib_msg_t    *cookie,
1015                  ptl_hdr_t    *hdr, 
1016                  int           type, 
1017                  ptl_nid_t     nid, 
1018                  ptl_pid_t     pid,
1019                  unsigned int  payload_niov, 
1020                  struct iovec *payload_iov, 
1021                  ptl_kiov_t   *payload_kiov,
1022                  size_t        payload_offset,
1023                  size_t        payload_nob)
1024 {
1025         ksock_ltx_t  *ltx;
1026         int           desc_size;
1027         int           rc;
1028
1029         /* NB 'private' is different depending on what we're sending.
1030          * Just ignore it... */
1031
1032         CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid:"LPX64
1033                " pid %d\n", payload_nob, payload_niov, nid , pid);
1034
1035         LASSERT (payload_nob == 0 || payload_niov > 0);
1036         LASSERT (payload_niov <= PTL_MD_MAX_IOV);
1037
1038         /* It must be OK to kmap() if required */
1039         LASSERT (payload_kiov == NULL || !in_interrupt ());
1040         /* payload is either all vaddrs or all pages */
1041         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
1042         
1043         if (payload_iov != NULL)
1044                 desc_size = offsetof(ksock_ltx_t, ltx_iov[1 + payload_niov]);
1045         else
1046                 desc_size = offsetof(ksock_ltx_t, ltx_kiov[payload_niov]);
1047         
1048         if (in_interrupt() ||
1049             type == PTL_MSG_ACK ||
1050             type == PTL_MSG_REPLY) {
1051                 /* Can't block if in interrupt or responding to an incoming
1052                  * message */
1053                 PORTAL_ALLOC_ATOMIC(ltx, desc_size);
1054         } else {
1055                 PORTAL_ALLOC(ltx, desc_size);
1056         }
1057         
1058         if (ltx == NULL) {
1059                 CERROR("Can't allocate tx desc type %d size %d %s\n",
1060                        type, desc_size, in_interrupt() ? "(intr)" : "");
1061                 return (PTL_NOSPACE);
1062         }
1063
1064         atomic_inc(&ksocknal_data.ksnd_nactive_ltxs);
1065
1066         ltx->ltx_desc_size = desc_size;
1067         
1068         /* We always have 1 mapped frag for the header */
1069         ltx->ltx_tx.tx_iov = ltx->ltx_iov;
1070         ltx->ltx_iov[0].iov_base = &ltx->ltx_hdr;
1071         ltx->ltx_iov[0].iov_len = sizeof(*hdr);
1072         ltx->ltx_hdr = *hdr;
1073         
1074         ltx->ltx_private = private;
1075         ltx->ltx_cookie = cookie;
1076         
1077         ltx->ltx_tx.tx_isfwd = 0;
1078         ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_nob;
1079
1080         if (payload_iov != NULL) {
1081                 /* payload is all mapped */
1082                 ltx->ltx_tx.tx_kiov  = NULL;
1083                 ltx->ltx_tx.tx_nkiov = 0;
1084
1085                 ltx->ltx_tx.tx_niov = 
1086                         1 + lib_extract_iov(payload_niov, &ltx->ltx_iov[1],
1087                                             payload_niov, payload_iov,
1088                                             payload_offset, payload_nob);
1089         } else {
1090                 /* payload is all pages */
1091                 ltx->ltx_tx.tx_niov = 1;
1092
1093                 ltx->ltx_tx.tx_kiov = ltx->ltx_kiov;
1094                 ltx->ltx_tx.tx_nkiov =
1095                         lib_extract_kiov(payload_niov, ltx->ltx_kiov,
1096                                          payload_niov, payload_kiov,
1097                                          payload_offset, payload_nob);
1098         }
1099
1100         rc = ksocknal_launch_packet(&ltx->ltx_tx, nid);
1101         if (rc == 0)
1102                 return (PTL_OK);
1103         
1104         ksocknal_free_ltx(ltx);
1105         return (PTL_FAIL);
1106 }
1107
1108 ptl_err_t
1109 ksocknal_send (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1110                ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1111                unsigned int payload_niov, struct iovec *payload_iov,
1112                size_t payload_offset, size_t payload_len)
1113 {
1114         return (ksocknal_sendmsg(nal, private, cookie,
1115                                  hdr, type, nid, pid,
1116                                  payload_niov, payload_iov, NULL,
1117                                  payload_offset, payload_len));
1118 }
1119
1120 ptl_err_t
1121 ksocknal_send_pages (nal_cb_t *nal, void *private, lib_msg_t *cookie, 
1122                      ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1123                      unsigned int payload_niov, ptl_kiov_t *payload_kiov, 
1124                      size_t payload_offset, size_t payload_len)
1125 {
1126         return (ksocknal_sendmsg(nal, private, cookie,
1127                                  hdr, type, nid, pid,
1128                                  payload_niov, NULL, payload_kiov,
1129                                  payload_offset, payload_len));
1130 }
1131
1132 void
1133 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1134 {
1135         ptl_nid_t     nid = fwd->kprfd_gateway_nid;
1136         ksock_ftx_t  *ftx = (ksock_ftx_t *)&fwd->kprfd_scratch;
1137         int           rc;
1138         
1139         CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1140                 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1141
1142         /* I'm the gateway; must be the last hop */
1143         if (nid == ksocknal_lib.ni.nid)
1144                 nid = fwd->kprfd_target_nid;
1145
1146         /* setup iov for hdr */
1147         ftx->ftx_iov.iov_base = fwd->kprfd_hdr;
1148         ftx->ftx_iov.iov_len = sizeof(ptl_hdr_t);
1149
1150         ftx->ftx_tx.tx_isfwd = 1;                  /* This is a forwarding packet */
1151         ftx->ftx_tx.tx_nob   = sizeof(ptl_hdr_t) + fwd->kprfd_nob;
1152         ftx->ftx_tx.tx_niov  = 1;
1153         ftx->ftx_tx.tx_iov   = &ftx->ftx_iov;
1154         ftx->ftx_tx.tx_nkiov = fwd->kprfd_niov;
1155         ftx->ftx_tx.tx_kiov  = fwd->kprfd_kiov;
1156
1157         rc = ksocknal_launch_packet (&ftx->ftx_tx, nid);
1158         if (rc != 0)
1159                 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
1160 }
1161
1162 int
1163 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1164 {
1165         long    pid = kernel_thread (fn, arg, 0);
1166
1167         if (pid < 0)
1168                 return ((int)pid);
1169
1170         atomic_inc (&ksocknal_data.ksnd_nthreads);
1171         return (0);
1172 }
1173
1174 void
1175 ksocknal_thread_fini (void)
1176 {
1177         atomic_dec (&ksocknal_data.ksnd_nthreads);
1178 }
1179
1180 void
1181 ksocknal_fmb_callback (void *arg, int error)
1182 {
1183         ksock_fmb_t       *fmb = (ksock_fmb_t *)arg;
1184         ksock_fmb_pool_t  *fmp = fmb->fmb_pool;
1185         ptl_hdr_t         *hdr = (ptl_hdr_t *)page_address(fmb->fmb_kiov[0].kiov_page);
1186         ksock_conn_t      *conn = NULL;
1187         ksock_sched_t     *sched;
1188         unsigned long      flags;
1189         char               ipbuf[PTL_NALFMT_SIZE];
1190         char               ipbuf2[PTL_NALFMT_SIZE];
1191
1192         if (error != 0)
1193                 CERROR("Failed to route packet from "
1194                        LPX64" %s to "LPX64" %s: %d\n",
1195                        NTOH__u64(hdr->src_nid),
1196                        portals_nid2str(SOCKNAL, NTOH__u64(hdr->src_nid), ipbuf),
1197                        NTOH__u64(hdr->dest_nid),
1198                        portals_nid2str(SOCKNAL, NTOH__u64(hdr->dest_nid), ipbuf2),
1199                        error);
1200         else
1201                 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1202                         NTOH__u64 (hdr->src_nid), NTOH__u64 (hdr->dest_nid));
1203
1204         /* drop peer ref taken on init */
1205         ksocknal_put_peer (fmb->fmb_peer);
1206
1207         spin_lock_irqsave (&fmp->fmp_lock, flags);
1208
1209         list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1210         fmp->fmp_nactive_fmbs--;
1211
1212         if (!list_empty (&fmp->fmp_blocked_conns)) {
1213                 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1214                                    ksock_conn_t, ksnc_rx_list);
1215                 list_del (&conn->ksnc_rx_list);
1216         }
1217
1218         spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1219
1220         if (conn == NULL)
1221                 return;
1222
1223         CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1224         LASSERT (conn->ksnc_rx_scheduled);
1225         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1226
1227         conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1228
1229         sched = conn->ksnc_scheduler;
1230
1231         spin_lock_irqsave (&sched->kss_lock, flags);
1232
1233         list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1234         wake_up (&sched->kss_waitq);
1235
1236         spin_unlock_irqrestore (&sched->kss_lock, flags);
1237 }
1238
1239 ksock_fmb_t *
1240 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1241 {
1242         int               payload_nob = conn->ksnc_rx_nob_left;
1243         unsigned long     flags;
1244         ksock_fmb_pool_t *pool;
1245         ksock_fmb_t      *fmb;
1246
1247         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1248         LASSERT (kpr_routing(&ksocknal_data.ksnd_router));
1249
1250         if (payload_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1251                 pool = &ksocknal_data.ksnd_small_fmp;
1252         else
1253                 pool = &ksocknal_data.ksnd_large_fmp;
1254
1255         spin_lock_irqsave (&pool->fmp_lock, flags);
1256
1257         if (!list_empty (&pool->fmp_idle_fmbs)) {
1258                 fmb = list_entry(pool->fmp_idle_fmbs.next,
1259                                  ksock_fmb_t, fmb_list);
1260                 list_del (&fmb->fmb_list);
1261                 pool->fmp_nactive_fmbs++;
1262                 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1263
1264                 return (fmb);
1265         }
1266
1267         /* deschedule until fmb free */
1268
1269         conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1270
1271         list_add_tail (&conn->ksnc_rx_list,
1272                        &pool->fmp_blocked_conns);
1273
1274         spin_unlock_irqrestore (&pool->fmp_lock, flags);
1275         return (NULL);
1276 }
1277
1278 int
1279 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1280 {
1281         int       payload_nob = conn->ksnc_rx_nob_left;
1282         ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1283         int       niov = 0;
1284         int       nob = payload_nob;
1285
1286         LASSERT (conn->ksnc_rx_scheduled);
1287         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1288         LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1289         LASSERT (payload_nob >= 0);
1290         LASSERT (payload_nob <= fmb->fmb_pool->fmp_buff_pages * PAGE_SIZE);
1291         LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1292         LASSERT (fmb->fmb_kiov[0].kiov_offset == 0);
1293
1294         /* Take a ref on the conn's peer to prevent module unload before
1295          * forwarding completes. */
1296         fmb->fmb_peer = conn->ksnc_peer;
1297         atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1298
1299         /* Copy the header we just read into the forwarding buffer.  If
1300          * there's payload, start reading reading it into the buffer,
1301          * otherwise the forwarding buffer can be kicked off
1302          * immediately. */
1303         fmb->fmb_hdr = conn->ksnc_hdr;
1304
1305         while (nob > 0) {
1306                 LASSERT (niov < fmb->fmb_pool->fmp_buff_pages);
1307                 LASSERT (fmb->fmb_kiov[niov].kiov_offset == 0);
1308                 fmb->fmb_kiov[niov].kiov_len = MIN (PAGE_SIZE, nob);
1309                 nob -= PAGE_SIZE;
1310                 niov++;
1311         }
1312
1313         kpr_fwd_init(&fmb->fmb_fwd, dest_nid, &fmb->fmb_hdr,
1314                      payload_nob, niov, fmb->fmb_kiov,
1315                      ksocknal_fmb_callback, fmb);
1316
1317         if (payload_nob == 0) {         /* got complete packet already */
1318                 CDEBUG (D_NET, "%p "LPX64"->"LPX64" fwd_start (immediate)\n",
1319                         conn, NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid);
1320
1321                 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1322
1323                 ksocknal_new_packet (conn, 0);  /* on to next packet */
1324                 return (1);
1325         }
1326
1327         conn->ksnc_cookie = fmb;                /* stash fmb for later */
1328         conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1329         
1330         /* Set up conn->ksnc_rx_kiov to read the payload into fmb's kiov-ed
1331          * buffer */
1332         LASSERT (niov <= sizeof(conn->ksnc_rx_iov_space)/sizeof(ptl_kiov_t));
1333
1334         conn->ksnc_rx_niov = 0;
1335         conn->ksnc_rx_nkiov = niov;
1336         conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1337         memcpy(conn->ksnc_rx_kiov, fmb->fmb_kiov, niov * sizeof(ptl_kiov_t));
1338         
1339         CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1340                 NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1341         return (0);
1342 }
1343
1344 void
1345 ksocknal_fwd_parse (ksock_conn_t *conn)
1346 {
1347         ksock_peer_t *peer;
1348         ptl_nid_t     dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1349         ptl_nid_t     src_nid = NTOH__u64 (conn->ksnc_hdr.src_nid);
1350         int           body_len = NTOH__u32 (conn->ksnc_hdr.payload_length);
1351         char str[PTL_NALFMT_SIZE];
1352         char str2[PTL_NALFMT_SIZE];
1353
1354         CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1355                 src_nid, dest_nid, conn->ksnc_rx_nob_left);
1356
1357         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1358         LASSERT (conn->ksnc_rx_scheduled);
1359
1360         if (body_len < 0) {                 /* length corrupt (overflow) */
1361                 CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
1362                        "packet size %d illegal\n",
1363                        src_nid, portals_nid2str(TCPNAL, src_nid, str),
1364                        dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1365                        body_len);
1366
1367                 ksocknal_new_packet (conn, 0);  /* on to new packet */
1368                 return;
1369         }
1370
1371         if (!kpr_routing(&ksocknal_data.ksnd_router)) {    /* not forwarding */
1372                 CERROR("dropping packet from "LPX64" (%s) for "LPX64
1373                        " (%s): not forwarding\n",
1374                        src_nid, portals_nid2str(TCPNAL, src_nid, str),
1375                        dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1376                 /* on to new packet (skip this one's body) */
1377                 ksocknal_new_packet (conn, body_len);
1378                 return;
1379         }
1380
1381         if (body_len > PTL_MTU) {      /* too big to forward */
1382                 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1383                         "(%s): packet size %d too big\n",
1384                         src_nid, portals_nid2str(TCPNAL, src_nid, str),
1385                         dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1386                         body_len);
1387                 /* on to new packet (skip this one's body) */
1388                 ksocknal_new_packet (conn, body_len);
1389                 return;
1390         }
1391
1392         /* should have gone direct */
1393         peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1394         if (peer != NULL) {
1395                 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1396                         "(%s): target is a peer\n",
1397                         src_nid, portals_nid2str(TCPNAL, src_nid, str),
1398                         dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1399                 ksocknal_put_peer (peer);  /* drop ref from get above */
1400
1401                 /* on to next packet (skip this one's body) */
1402                 ksocknal_new_packet (conn, body_len);
1403                 return;
1404         }
1405
1406         conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;       /* Getting FMB now */
1407         conn->ksnc_rx_nob_left = body_len;              /* stash packet size */
1408         conn->ksnc_rx_nob_wanted = body_len;            /* (no slop) */
1409 }
1410
1411 int
1412 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1413 {
1414         static char ksocknal_slop_buffer[4096];
1415
1416         int   nob;
1417         int   niov;
1418         int   skipped;
1419
1420         if (nob_to_skip == 0) {         /* right at next packet boundary now */
1421                 conn->ksnc_rx_started = 0;
1422                 mb ();                          /* racing with timeout thread */
1423                 
1424                 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1425                 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1426                 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1427
1428                 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1429                 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1430                 conn->ksnc_rx_iov[0].iov_len  = sizeof (ptl_hdr_t);
1431                 conn->ksnc_rx_niov = 1;
1432
1433                 conn->ksnc_rx_kiov = NULL;
1434                 conn->ksnc_rx_nkiov = 0;
1435                 return (1);
1436         }
1437
1438         /* Set up to skip as much a possible now.  If there's more left
1439          * (ran out of iov entries) we'll get called again */
1440
1441         conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1442         conn->ksnc_rx_nob_left = nob_to_skip;
1443         conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1444         skipped = 0;
1445         niov = 0;
1446
1447         do {
1448                 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1449
1450                 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1451                 conn->ksnc_rx_iov[niov].iov_len  = nob;
1452                 niov++;
1453                 skipped += nob;
1454                 nob_to_skip -=nob;
1455
1456         } while (nob_to_skip != 0 &&    /* mustn't overflow conn's rx iov */
1457                  niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1458
1459         conn->ksnc_rx_niov = niov;
1460         conn->ksnc_rx_kiov = NULL;
1461         conn->ksnc_rx_nkiov = 0;
1462         conn->ksnc_rx_nob_wanted = skipped;
1463         return (0);
1464 }
1465
1466 int
1467 ksocknal_process_receive (ksock_conn_t *conn)
1468 {
1469         ksock_fmb_t  *fmb;
1470         int           rc;
1471         
1472         LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1473
1474         /* doesn't need a forwarding buffer */
1475         if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1476                 goto try_read;
1477
1478  get_fmb:
1479         fmb = ksocknal_get_idle_fmb (conn);
1480         if (fmb == NULL) {
1481                 /* conn descheduled waiting for idle fmb */
1482                 return (0);
1483         }
1484
1485         if (ksocknal_init_fmb (conn, fmb)) {
1486                 /* packet forwarded */
1487                 return (0);
1488         }
1489
1490  try_read:
1491         /* NB: sched lock NOT held */
1492         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1493                  conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1494                  conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1495                  conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1496
1497         LASSERT (conn->ksnc_rx_nob_wanted > 0);
1498
1499         rc = ksocknal_receive(conn);
1500
1501         if (rc <= 0) {
1502                 LASSERT (rc != -EAGAIN);
1503
1504                 if (rc == 0)
1505                         CWARN ("[%p] EOF from "LPX64" ip %d.%d.%d.%d:%d\n",
1506                                conn, conn->ksnc_peer->ksnp_nid,
1507                                HIPQUAD(conn->ksnc_ipaddr),
1508                                conn->ksnc_port);
1509                 else if (!conn->ksnc_closing)
1510                         CERROR ("[%p] Error %d on read from "LPX64
1511                                 " ip %d.%d.%d.%d:%d\n",
1512                                 conn, rc, conn->ksnc_peer->ksnp_nid,
1513                                 HIPQUAD(conn->ksnc_ipaddr),
1514                                 conn->ksnc_port);
1515
1516                 ksocknal_close_conn_and_siblings (conn, rc);
1517                 return (rc == 0 ? -ESHUTDOWN : rc);
1518         }
1519
1520         if (conn->ksnc_rx_nob_wanted != 0) {
1521                 /* short read */
1522                 return (-EAGAIN);
1523         }
1524         
1525         switch (conn->ksnc_rx_state) {
1526         case SOCKNAL_RX_HEADER:
1527                 if (conn->ksnc_hdr.type != HTON__u32(PTL_MSG_HELLO) &&
1528                     NTOH__u64(conn->ksnc_hdr.dest_nid) != ksocknal_lib.ni.nid) {
1529                         /* This packet isn't for me */
1530                         ksocknal_fwd_parse (conn);
1531                         switch (conn->ksnc_rx_state) {
1532                         case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1533                                 return (0);     /* => come back later */
1534                         case SOCKNAL_RX_SLOP:   /* skipping packet's body */
1535                                 goto try_read;  /* => go read it */
1536                         case SOCKNAL_RX_GET_FMB: /* forwarding */
1537                                 goto get_fmb;   /* => go get a fwd msg buffer */
1538                         default:
1539                                 LBUG ();
1540                         }
1541                         /* Not Reached */
1542                 }
1543
1544                 /* sets wanted_len, iovs etc */
1545                 lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1546
1547                 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1548                         conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1549                         goto try_read;          /* go read the payload */
1550                 }
1551                 /* Fall through (completed packet for me) */
1552
1553         case SOCKNAL_RX_BODY:
1554                 /* payload all received */
1555                 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_OK);
1556                 /* Fall through */
1557
1558         case SOCKNAL_RX_SLOP:
1559                 /* starting new packet? */
1560                 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1561                         return (0);     /* come back later */
1562                 goto try_read;          /* try to finish reading slop now */
1563
1564         case SOCKNAL_RX_BODY_FWD:
1565                 /* payload all received */
1566                 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1567                         conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1568                         NTOH__u64 (conn->ksnc_hdr.dest_nid),
1569                         conn->ksnc_rx_nob_left);
1570
1571                 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1572                  * conn->ksnc_cookie */
1573                 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1574                 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1575
1576                 /* no slop in forwarded packets */
1577                 LASSERT (conn->ksnc_rx_nob_left == 0);
1578
1579                 ksocknal_new_packet (conn, 0);  /* on to next packet */
1580                 return (0);                     /* (later) */
1581
1582         default:
1583                 break;
1584         }
1585
1586         /* Not Reached */
1587         LBUG ();
1588         return (-EINVAL);                       /* keep gcc happy */
1589 }
1590
1591 ptl_err_t
1592 ksocknal_recv (nal_cb_t *nal, void *private, lib_msg_t *msg,
1593                unsigned int niov, struct iovec *iov, 
1594                size_t offset, size_t mlen, size_t rlen)
1595 {
1596         ksock_conn_t *conn = (ksock_conn_t *)private;
1597
1598         LASSERT (mlen <= rlen);
1599         LASSERT (niov <= PTL_MD_MAX_IOV);
1600         
1601         conn->ksnc_cookie = msg;
1602         conn->ksnc_rx_nob_wanted = mlen;
1603         conn->ksnc_rx_nob_left   = rlen;
1604
1605         conn->ksnc_rx_nkiov = 0;
1606         conn->ksnc_rx_kiov = NULL;
1607         conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1608         conn->ksnc_rx_niov =
1609                 lib_extract_iov(PTL_MD_MAX_IOV, conn->ksnc_rx_iov,
1610                                 niov, iov, offset, mlen);
1611
1612         LASSERT (mlen == 
1613                  lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1614                  lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1615
1616         return (PTL_OK);
1617 }
1618
1619 ptl_err_t
1620 ksocknal_recv_pages (nal_cb_t *nal, void *private, lib_msg_t *msg,
1621                      unsigned int niov, ptl_kiov_t *kiov, 
1622                      size_t offset, size_t mlen, size_t rlen)
1623 {
1624         ksock_conn_t *conn = (ksock_conn_t *)private;
1625
1626         LASSERT (mlen <= rlen);
1627         LASSERT (niov <= PTL_MD_MAX_IOV);
1628         
1629         conn->ksnc_cookie = msg;
1630         conn->ksnc_rx_nob_wanted = mlen;
1631         conn->ksnc_rx_nob_left   = rlen;
1632
1633         conn->ksnc_rx_niov = 0;
1634         conn->ksnc_rx_iov  = NULL;
1635         conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1636         conn->ksnc_rx_nkiov = 
1637                 lib_extract_kiov(PTL_MD_MAX_IOV, conn->ksnc_rx_kiov,
1638                                  niov, kiov, offset, mlen);
1639
1640         LASSERT (mlen == 
1641                  lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1642                  lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1643
1644         return (PTL_OK);
1645 }
1646
1647 int ksocknal_scheduler (void *arg)
1648 {
1649         ksock_sched_t     *sched = (ksock_sched_t *)arg;
1650         ksock_conn_t      *conn;
1651         ksock_tx_t        *tx;
1652         unsigned long      flags;
1653         int                rc;
1654         int                nloops = 0;
1655         int                id = sched - ksocknal_data.ksnd_schedulers;
1656         char               name[16];
1657
1658         snprintf (name, sizeof (name),"ksocknald_%02d", id);
1659         kportal_daemonize (name);
1660         kportal_blockallsigs ();
1661
1662 #if (CONFIG_SMP && CPU_AFFINITY)
1663         if ((cpu_online_map & (1 << id)) != 0) {
1664 #if 1
1665                 current->cpus_allowed = (1 << id);
1666 #else
1667                 set_cpus_allowed (current, 1<<id);
1668 #endif
1669         } else {
1670                 CERROR ("Can't set CPU affinity for %s\n", name);
1671         }
1672 #endif /* CONFIG_SMP && CPU_AFFINITY */
1673         
1674         spin_lock_irqsave (&sched->kss_lock, flags);
1675
1676         while (!ksocknal_data.ksnd_shuttingdown) {
1677                 int did_something = 0;
1678
1679                 /* Ensure I progress everything semi-fairly */
1680
1681                 if (!list_empty (&sched->kss_rx_conns)) {
1682                         conn = list_entry(sched->kss_rx_conns.next,
1683                                           ksock_conn_t, ksnc_rx_list);
1684                         list_del(&conn->ksnc_rx_list);
1685
1686                         LASSERT(conn->ksnc_rx_scheduled);
1687                         LASSERT(conn->ksnc_rx_ready);
1688
1689                         /* clear rx_ready in case receive isn't complete.
1690                          * Do it BEFORE we call process_recv, since
1691                          * data_ready can set it any time after we release
1692                          * kss_lock. */
1693                         conn->ksnc_rx_ready = 0;
1694                         spin_unlock_irqrestore(&sched->kss_lock, flags);
1695                         
1696                         rc = ksocknal_process_receive(conn);
1697                         
1698                         spin_lock_irqsave(&sched->kss_lock, flags);
1699
1700                         /* I'm the only one that can clear this flag */
1701                         LASSERT(conn->ksnc_rx_scheduled);
1702
1703                         /* Did process_receive get everything it wanted? */
1704                         if (rc == 0)
1705                                 conn->ksnc_rx_ready = 1;
1706                         
1707                         if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
1708                             conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
1709                                 /* Conn blocked for a forwarding buffer.
1710                                  * It will get queued for my attention when
1711                                  * one becomes available (and it might just
1712                                  * already have been!).  Meanwhile my ref
1713                                  * on it stays put. */
1714                         } else if (conn->ksnc_rx_ready) {
1715                                 /* reschedule for rx */
1716                                 list_add_tail (&conn->ksnc_rx_list,
1717                                                &sched->kss_rx_conns);
1718                         } else {
1719                                 conn->ksnc_rx_scheduled = 0;
1720                                 /* drop my ref */
1721                                 ksocknal_put_conn(conn);
1722                         }
1723
1724                         did_something = 1;
1725                 }
1726
1727                 if (!list_empty (&sched->kss_tx_conns)) {
1728                         conn = list_entry(sched->kss_tx_conns.next,
1729                                           ksock_conn_t, ksnc_tx_list);
1730                         list_del (&conn->ksnc_tx_list);
1731                         
1732                         LASSERT(conn->ksnc_tx_scheduled);
1733                         LASSERT(conn->ksnc_tx_ready);
1734                         LASSERT(!list_empty(&conn->ksnc_tx_queue));
1735                         
1736                         tx = list_entry(conn->ksnc_tx_queue.next,
1737                                         ksock_tx_t, tx_list);
1738                         /* dequeue now so empty list => more to send */
1739                         list_del(&tx->tx_list);
1740                         
1741                         /* Clear tx_ready in case send isn't complete.  Do
1742                          * it BEFORE we call process_transmit, since
1743                          * write_space can set it any time after we release
1744                          * kss_lock. */
1745                         conn->ksnc_tx_ready = 0;
1746                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1747
1748                         rc = ksocknal_process_transmit(conn, tx);
1749
1750                         spin_lock_irqsave (&sched->kss_lock, flags);
1751
1752                         if (rc == -ENOMEM || rc == -EAGAIN) {
1753                                 /* Incomplete send: replace tx on HEAD of tx_queue */
1754                                 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
1755                         } else {
1756                                 /* Complete send; assume space for more */
1757                                 conn->ksnc_tx_ready = 1;
1758                         }
1759
1760                         if (rc == -ENOMEM) {
1761                                 /* Do nothing; after a short timeout, this
1762                                  * conn will be reposted on kss_tx_conns. */
1763                         } else if (conn->ksnc_tx_ready &&
1764                                    !list_empty (&conn->ksnc_tx_queue)) {
1765                                 /* reschedule for tx */
1766                                 list_add_tail (&conn->ksnc_tx_list, 
1767                                                &sched->kss_tx_conns);
1768                         } else {
1769                                 conn->ksnc_tx_scheduled = 0;
1770                                 /* drop my ref */
1771                                 ksocknal_put_conn (conn);
1772                         }
1773                                 
1774                         did_something = 1;
1775                 }
1776 #if SOCKNAL_ZC
1777                 if (!list_empty (&sched->kss_zctxdone_list)) {
1778                         ksock_tx_t *tx =
1779                                 list_entry(sched->kss_zctxdone_list.next,
1780                                            ksock_tx_t, tx_list);
1781                         did_something = 1;
1782
1783                         list_del (&tx->tx_list);
1784                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1785
1786                         ksocknal_tx_done (tx, 1);
1787
1788                         spin_lock_irqsave (&sched->kss_lock, flags);
1789                 }
1790 #endif
1791                 if (!did_something ||           /* nothing to do */
1792                     ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1793                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1794
1795                         nloops = 0;
1796
1797                         if (!did_something) {   /* wait for something to do */
1798 #if SOCKNAL_ZC
1799                                 rc = wait_event_interruptible (sched->kss_waitq,
1800                                                                ksocknal_data.ksnd_shuttingdown ||
1801                                                                !list_empty(&sched->kss_rx_conns) ||
1802                                                                !list_empty(&sched->kss_tx_conns) ||
1803                                                                !list_empty(&sched->kss_zctxdone_list));
1804 #else
1805                                 rc = wait_event_interruptible (sched->kss_waitq,
1806                                                                ksocknal_data.ksnd_shuttingdown ||
1807                                                                !list_empty(&sched->kss_rx_conns) ||
1808                                                                !list_empty(&sched->kss_tx_conns));
1809 #endif
1810                                 LASSERT (rc == 0);
1811                         } else
1812                                our_cond_resched();
1813
1814                         spin_lock_irqsave (&sched->kss_lock, flags);
1815                 }
1816         }
1817
1818         spin_unlock_irqrestore (&sched->kss_lock, flags);
1819         ksocknal_thread_fini ();
1820         return (0);
1821 }
1822
1823 void
1824 ksocknal_data_ready (struct sock *sk, int n)
1825 {
1826         unsigned long  flags;
1827         ksock_conn_t  *conn;
1828         ksock_sched_t *sched;
1829         ENTRY;
1830
1831         /* interleave correctly with closing sockets... */
1832         read_lock (&ksocknal_data.ksnd_global_lock);
1833
1834         conn = sk->sk_user_data;
1835         if (conn == NULL) {             /* raced with ksocknal_terminate_conn */
1836                 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1837                 sk->sk_data_ready (sk, n);
1838         } else {
1839                 sched = conn->ksnc_scheduler;
1840
1841                 spin_lock_irqsave (&sched->kss_lock, flags);
1842
1843                 conn->ksnc_rx_ready = 1;
1844
1845                 if (!conn->ksnc_rx_scheduled) {  /* not being progressed */
1846                         list_add_tail(&conn->ksnc_rx_list,
1847                                       &sched->kss_rx_conns);
1848                         conn->ksnc_rx_scheduled = 1;
1849                         /* extra ref for scheduler */
1850                         atomic_inc (&conn->ksnc_refcount);
1851
1852                         wake_up (&sched->kss_waitq);
1853                 }
1854
1855                 spin_unlock_irqrestore (&sched->kss_lock, flags);
1856         }
1857
1858         read_unlock (&ksocknal_data.ksnd_global_lock);
1859
1860         EXIT;
1861 }
1862
1863 void
1864 ksocknal_write_space (struct sock *sk)
1865 {
1866         unsigned long  flags;
1867         ksock_conn_t  *conn;
1868         ksock_sched_t *sched;
1869
1870         /* interleave correctly with closing sockets... */
1871         read_lock (&ksocknal_data.ksnd_global_lock);
1872
1873         conn = sk->sk_user_data;
1874
1875         CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1876                sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1877                (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1878                                       " ready" : " blocked"),
1879                (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1880                                       " scheduled" : " idle"),
1881                (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1882                                       " empty" : " queued"));
1883
1884         if (conn == NULL) {             /* raced with ksocknal_terminate_conn */
1885                 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1886                 sk->sk_write_space (sk);
1887
1888                 read_unlock (&ksocknal_data.ksnd_global_lock);
1889                 return;
1890         }
1891
1892         if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1893                 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1894
1895                 sched = conn->ksnc_scheduler;
1896
1897                 spin_lock_irqsave (&sched->kss_lock, flags);
1898
1899                 conn->ksnc_tx_ready = 1;
1900
1901                 if (!conn->ksnc_tx_scheduled && // not being progressed
1902                     !list_empty(&conn->ksnc_tx_queue)){//packets to send
1903                         list_add_tail (&conn->ksnc_tx_list,
1904                                        &sched->kss_tx_conns);
1905                         conn->ksnc_tx_scheduled = 1;
1906                         /* extra ref for scheduler */
1907                         atomic_inc (&conn->ksnc_refcount);
1908
1909                         wake_up (&sched->kss_waitq);
1910                 }
1911
1912                 spin_unlock_irqrestore (&sched->kss_lock, flags);
1913         }
1914
1915         read_unlock (&ksocknal_data.ksnd_global_lock);
1916 }
1917
1918 int
1919 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1920 {
1921         int           rc;
1922         mm_segment_t  oldmm = get_fs();
1923
1924         while (nob > 0) {
1925                 struct iovec  iov = {
1926                         .iov_base = buffer,
1927                         .iov_len  = nob
1928                 };
1929                 struct msghdr msg = {
1930                         .msg_name       = NULL,
1931                         .msg_namelen    = 0,
1932                         .msg_iov        = &iov,
1933                         .msg_iovlen     = 1,
1934                         .msg_control    = NULL,
1935                         .msg_controllen = 0,
1936                         .msg_flags      = 0
1937                 };
1938
1939                 set_fs (KERNEL_DS);
1940                 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1941                 set_fs (oldmm);
1942                 
1943                 if (rc < 0)
1944                         return (rc);
1945
1946                 if (rc == 0) {
1947                         CERROR ("Unexpected zero rc\n");
1948                         return (-ECONNABORTED);
1949                 }
1950
1951                 buffer = ((char *)buffer) + rc;
1952                 nob -= rc;
1953         }
1954         
1955         return (0);
1956 }
1957
1958 int
1959 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1960 {
1961         int           rc;
1962         mm_segment_t  oldmm = get_fs();
1963         
1964         while (nob > 0) {
1965                 struct iovec  iov = {
1966                         .iov_base = buffer,
1967                         .iov_len  = nob
1968                 };
1969                 struct msghdr msg = {
1970                         .msg_name       = NULL,
1971                         .msg_namelen    = 0,
1972                         .msg_iov        = &iov,
1973                         .msg_iovlen     = 1,
1974                         .msg_control    = NULL,
1975                         .msg_controllen = 0,
1976                         .msg_flags      = 0
1977                 };
1978
1979                 set_fs (KERNEL_DS);
1980                 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
1981                 set_fs (oldmm);
1982                 
1983                 if (rc < 0)
1984                         return (rc);
1985
1986                 if (rc == 0)
1987                         return (-ECONNABORTED);
1988
1989                 buffer = ((char *)buffer) + rc;
1990                 nob -= rc;
1991         }
1992         
1993         return (0);
1994 }
1995
1996 int
1997 ksocknal_hello (struct socket *sock, ptl_nid_t *nid, int *type,
1998                 __u64 *incarnation)
1999 {
2000         int                 rc;
2001         ptl_hdr_t           hdr;
2002         ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
2003         char                ipbuf[PTL_NALFMT_SIZE];
2004         char                ipbuf2[PTL_NALFMT_SIZE];
2005
2006         LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
2007
2008         memset (&hdr, 0, sizeof (hdr));
2009         hmv->magic         = __cpu_to_le32 (PORTALS_PROTO_MAGIC);
2010         hmv->version_major = __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR);
2011         hmv->version_minor = __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR);
2012
2013         hdr.src_nid = __cpu_to_le64 (ksocknal_lib.ni.nid);
2014         hdr.type    = __cpu_to_le32 (PTL_MSG_HELLO);
2015
2016         hdr.msg.hello.type = __cpu_to_le32 (*type);
2017         hdr.msg.hello.incarnation =
2018                 __cpu_to_le64 (ksocknal_data.ksnd_incarnation);
2019
2020         /* Assume sufficient socket buffering for this message */
2021         rc = ksocknal_sock_write (sock, &hdr, sizeof (hdr));
2022         if (rc != 0) {
2023                 CERROR ("Error %d sending HELLO to "LPX64" %s\n",
2024                         rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2025                 return (rc);
2026         }
2027
2028         rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
2029         if (rc != 0) {
2030                 CERROR ("Error %d reading HELLO from "LPX64" %s\n",
2031                         rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2032                 return (rc);
2033         }
2034
2035         if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
2036                 CERROR ("Bad magic %#08x (%#08x expected) from "LPX64" %s\n",
2037                         __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, *nid,
2038                         portals_nid2str(SOCKNAL, *nid, ipbuf));
2039                 return (-EPROTO);
2040         }
2041
2042         if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
2043             hmv->version_minor != __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
2044                 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
2045                         " from "LPX64" %s\n",
2046                         __le16_to_cpu (hmv->version_major),
2047                         __le16_to_cpu (hmv->version_minor),
2048                         PORTALS_PROTO_VERSION_MAJOR,
2049                         PORTALS_PROTO_VERSION_MINOR,
2050                         *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2051                 return (-EPROTO);
2052         }
2053
2054 #if (PORTALS_PROTO_VERSION_MAJOR != 0)
2055 # error "This code only understands protocol version 0.x"
2056 #endif
2057         /* version 0 sends magic/version as the dest_nid of a 'hello' header,
2058          * so read the rest of it in now... */
2059
2060         rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
2061         if (rc != 0) {
2062                 CERROR ("Error %d reading rest of HELLO hdr from "LPX64" %s\n",
2063                         rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2064                 return (rc);
2065         }
2066
2067         /* ...and check we got what we expected */
2068         if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
2069             hdr.payload_length != __cpu_to_le32 (0)) {
2070                 CERROR ("Expecting a HELLO hdr with 0 payload,"
2071                         " but got type %d with %d payload from "LPX64" %s\n",
2072                         __le32_to_cpu (hdr.type),
2073                         __le32_to_cpu (hdr.payload_length), *nid,
2074                         portals_nid2str(SOCKNAL, *nid, ipbuf));
2075                 return (-EPROTO);
2076         }
2077
2078         if (__le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
2079                 CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY\n");
2080                 return (-EPROTO);
2081         }
2082
2083         if (*nid == PTL_NID_ANY) {              /* don't know peer's nid yet */
2084                 *nid = __le64_to_cpu(hdr.src_nid);
2085         } else if (*nid != __le64_to_cpu (hdr.src_nid)) {
2086                 CERROR ("Connected to nid "LPX64" %s, but expecting "LPX64" %s\n",
2087                         __le64_to_cpu (hdr.src_nid),
2088                         portals_nid2str(SOCKNAL,
2089                                         __le64_to_cpu(hdr.src_nid),
2090                                         ipbuf),
2091                         *nid, portals_nid2str(SOCKNAL, *nid, ipbuf2));
2092                 return (-EPROTO);
2093         }
2094
2095         if (*type == SOCKNAL_CONN_NONE) {
2096                 /* I've accepted this connection; peer determines type */
2097                 *type = __le32_to_cpu(hdr.msg.hello.type);
2098                 switch (*type) {
2099                 case SOCKNAL_CONN_ANY:
2100                 case SOCKNAL_CONN_CONTROL:
2101                         break;
2102                 case SOCKNAL_CONN_BULK_IN:
2103                         *type = SOCKNAL_CONN_BULK_OUT;
2104                         break;
2105                 case SOCKNAL_CONN_BULK_OUT:
2106                         *type = SOCKNAL_CONN_BULK_IN;
2107                         break;
2108                 default:
2109                         CERROR ("Unexpected type %d from "LPX64" %s\n",
2110                                 *type, *nid,
2111                                 portals_nid2str(SOCKNAL, *nid, ipbuf));
2112                         return (-EPROTO);
2113                 }
2114         } else if (__le32_to_cpu(hdr.msg.hello.type) != SOCKNAL_CONN_NONE) {
2115                 CERROR ("Mismatched types: me %d "LPX64" %s %d\n",
2116                         *type, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf),
2117                         __le32_to_cpu(hdr.msg.hello.type));
2118                 return (-EPROTO);
2119         }
2120
2121         *incarnation = __le64_to_cpu(hdr.msg.hello.incarnation);
2122
2123         return (0);
2124 }
2125
2126 int
2127 ksocknal_setup_sock (struct socket *sock)
2128 {
2129         mm_segment_t    oldmm = get_fs ();
2130         int             rc;
2131         int             option;
2132         struct linger   linger;
2133
2134 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
2135         sock->sk->sk_allocation = GFP_NOFS;
2136 #else
2137         sock->sk->allocation = GFP_NOFS;
2138 #endif
2139
2140         /* Ensure this socket aborts active sends immediately when we close
2141          * it. */
2142
2143         linger.l_onoff = 0;
2144         linger.l_linger = 0;
2145
2146         set_fs (KERNEL_DS);
2147         rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2148                               (char *)&linger, sizeof (linger));
2149         set_fs (oldmm);
2150         if (rc != 0) {
2151                 CERROR ("Can't set SO_LINGER: %d\n", rc);
2152                 return (rc);
2153         }
2154
2155         option = -1;
2156         set_fs (KERNEL_DS);
2157         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2158                                     (char *)&option, sizeof (option));
2159         set_fs (oldmm);
2160         if (rc != 0) {
2161                 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2162                 return (rc);
2163         }
2164
2165 #if SOCKNAL_USE_KEEPALIVES
2166         /* Keepalives: If 3/4 of the timeout elapses, start probing every
2167          * second until the timeout elapses. */
2168
2169         option = (ksocknal_data.ksnd_io_timeout * 3) / 4;
2170         set_fs (KERNEL_DS);
2171         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2172                                     (char *)&option, sizeof (option));
2173         set_fs (oldmm);
2174         if (rc != 0) {
2175                 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2176                 return (rc);
2177         }
2178         
2179         option = 1;
2180         set_fs (KERNEL_DS);
2181         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2182                                     (char *)&option, sizeof (option));
2183         set_fs (oldmm);
2184         if (rc != 0) {
2185                 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2186                 return (rc);
2187         }
2188         
2189         option = ksocknal_data.ksnd_io_timeout / 4;
2190         set_fs (KERNEL_DS);
2191         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2192                                     (char *)&option, sizeof (option));
2193         set_fs (oldmm);
2194         if (rc != 0) {
2195                 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2196                 return (rc);
2197         }
2198
2199         option = 1;
2200         set_fs (KERNEL_DS);
2201         rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE, 
2202                               (char *)&option, sizeof (option));
2203         set_fs (oldmm);
2204         if (rc != 0) {
2205                 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2206                 return (rc);
2207         }
2208 #endif
2209         return (0);
2210 }
2211
2212 int
2213 ksocknal_connect_peer (ksock_route_t *route, int type)
2214 {
2215         struct sockaddr_in  peer_addr;
2216         mm_segment_t        oldmm = get_fs();
2217         struct timeval      tv;
2218         int                 fd;
2219         struct socket      *sock;
2220         int                 rc;
2221         char                ipbuf[PTL_NALFMT_SIZE];
2222
2223         rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2224         if (rc != 0) {
2225                 CERROR ("Can't create autoconnect socket: %d\n", rc);
2226                 return (rc);
2227         }
2228
2229         /* Ugh; have to map_fd for compatibility with sockets passed in
2230          * from userspace.  And we actually need the sock->file refcounting
2231          * that this gives you :) */
2232
2233         fd = sock_map_fd (sock);
2234         if (fd < 0) {
2235                 sock_release (sock);
2236                 CERROR ("sock_map_fd error %d\n", fd);
2237                 return (fd);
2238         }
2239
2240         /* NB the fd now owns the ref on sock->file */
2241         LASSERT (sock->file != NULL);
2242         LASSERT (file_count(sock->file) == 1);
2243
2244         /* Set the socket timeouts, so our connection attempt completes in
2245          * finite time */
2246         tv.tv_sec = ksocknal_data.ksnd_io_timeout;
2247         tv.tv_usec = 0;
2248
2249         set_fs (KERNEL_DS);
2250         rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2251                               (char *)&tv, sizeof (tv));
2252         set_fs (oldmm);
2253         if (rc != 0) {
2254                 CERROR ("Can't set send timeout %d: %d\n", 
2255                         ksocknal_data.ksnd_io_timeout, rc);
2256                 goto out;
2257         }
2258         
2259         set_fs (KERNEL_DS);
2260         rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2261                               (char *)&tv, sizeof (tv));
2262         set_fs (oldmm);
2263         if (rc != 0) {
2264                 CERROR ("Can't set receive timeout %d: %d\n",
2265                         ksocknal_data.ksnd_io_timeout, rc);
2266                 goto out;
2267         }
2268
2269         {
2270                 int  option = 1;
2271                 
2272                 set_fs (KERNEL_DS);
2273                 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2274                                             (char *)&option, sizeof (option));
2275                 set_fs (oldmm);
2276                 if (rc != 0) {
2277                         CERROR ("Can't disable nagle: %d\n", rc);
2278                         goto out;
2279                 }
2280         }
2281         
2282         if (route->ksnr_buffer_size != 0) {
2283                 int option = route->ksnr_buffer_size;
2284                 
2285                 set_fs (KERNEL_DS);
2286                 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2287                                       (char *)&option, sizeof (option));
2288                 set_fs (oldmm);
2289                 if (rc != 0) {
2290                         CERROR ("Can't set send buffer %d: %d\n",
2291                                 route->ksnr_buffer_size, rc);
2292                         goto out;
2293                 }
2294
2295                 set_fs (KERNEL_DS);
2296                 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2297                                       (char *)&option, sizeof (option));
2298                 set_fs (oldmm);
2299                 if (rc != 0) {
2300                         CERROR ("Can't set receive buffer %d: %d\n",
2301                                 route->ksnr_buffer_size, rc);
2302                         goto out;
2303                 }
2304         }
2305         
2306         memset (&peer_addr, 0, sizeof (peer_addr));
2307         peer_addr.sin_family = AF_INET;
2308         peer_addr.sin_port = htons (route->ksnr_port);
2309         peer_addr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2310         
2311         rc = sock->ops->connect (sock, (struct sockaddr *)&peer_addr, 
2312                                  sizeof (peer_addr), sock->file->f_flags);
2313         if (rc != 0) {
2314                 CERROR ("Error %d connecting to "LPX64" %s\n", rc,
2315                         route->ksnr_peer->ksnp_nid,
2316                         portals_nid2str(SOCKNAL,
2317                                         route->ksnr_peer->ksnp_nid,
2318                                         ipbuf));
2319                 goto out;
2320         }
2321         
2322         rc = ksocknal_create_conn (route, sock, route->ksnr_irq_affinity, type);
2323         if (rc == 0) {
2324                 /* Take an extra ref on sock->file to compensate for the
2325                  * upcoming close which will lose fd's ref on it. */
2326                 get_file (sock->file);
2327         }
2328
2329  out:
2330         sys_close (fd);
2331         return (rc);
2332 }
2333
2334 void
2335 ksocknal_autoconnect (ksock_route_t *route)
2336 {
2337         LIST_HEAD        (zombies);
2338         ksock_tx_t       *tx;
2339         ksock_peer_t     *peer;
2340         unsigned long     flags;
2341         int               rc;
2342         int               type;
2343         
2344         for (;;) {
2345                 for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
2346                         if ((route->ksnr_connecting & (1 << type)) != 0)
2347                                 break;
2348                 LASSERT (type < SOCKNAL_CONN_NTYPES);
2349
2350                 rc = ksocknal_connect_peer (route, type);
2351
2352                 if (rc != 0)
2353                         break;
2354                 
2355                 /* successfully autoconnected: create_conn did the
2356                  * route/conn binding and scheduled any blocked packets */
2357
2358                 if (route->ksnr_connecting == 0) {
2359                         /* No more connections required */
2360                         return;
2361                 }
2362         }
2363
2364         /* Connection attempt failed */
2365
2366         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2367
2368         peer = route->ksnr_peer;
2369         route->ksnr_connecting = 0;
2370
2371         /* This is a retry rather than a new connection */
2372         LASSERT (route->ksnr_retry_interval != 0);
2373         route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2374         route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2375                                           SOCKNAL_MAX_RECONNECT_INTERVAL);
2376
2377         if (!list_empty (&peer->ksnp_tx_queue) &&
2378             ksocknal_find_connecting_route_locked (peer) == NULL) {
2379                 LASSERT (list_empty (&peer->ksnp_conns));
2380
2381                 /* None of the connections that the blocked packets are
2382                  * waiting for have been successful.  Complete them now... */
2383                 do {
2384                         tx = list_entry (peer->ksnp_tx_queue.next,
2385                                          ksock_tx_t, tx_list);
2386                         list_del (&tx->tx_list);
2387                         list_add_tail (&tx->tx_list, &zombies);
2388                 } while (!list_empty (&peer->ksnp_tx_queue));
2389         }
2390
2391         /* make this route least-favourite for re-selection */
2392         if (!route->ksnr_deleted) {
2393                 list_del(&route->ksnr_list);
2394                 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
2395         }
2396         
2397         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2398
2399         while (!list_empty (&zombies)) {
2400                 char ipbuf[PTL_NALFMT_SIZE];
2401                 char ipbuf2[PTL_NALFMT_SIZE];
2402                 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2403
2404                 CERROR ("Deleting packet type %d len %d ("LPX64" %s->"LPX64" %s)\n",
2405                         NTOH__u32 (tx->tx_hdr->type),
2406                         NTOH__u32 (tx->tx_hdr->payload_length),
2407                         NTOH__u64 (tx->tx_hdr->src_nid),
2408                         portals_nid2str(SOCKNAL,
2409                                         NTOH__u64(tx->tx_hdr->src_nid),
2410                                         ipbuf),
2411                         NTOH__u64 (tx->tx_hdr->dest_nid),
2412                         portals_nid2str(SOCKNAL,
2413                                         NTOH__u64(tx->tx_hdr->src_nid),
2414                                         ipbuf2));
2415
2416                 list_del (&tx->tx_list);
2417                 /* complete now */
2418                 ksocknal_tx_done (tx, 0);
2419         }
2420 }
2421
2422 int
2423 ksocknal_autoconnectd (void *arg)
2424 {
2425         long               id = (long)arg;
2426         char               name[16];
2427         unsigned long      flags;
2428         ksock_route_t     *route;
2429         int                rc;
2430
2431         snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2432         kportal_daemonize (name);
2433         kportal_blockallsigs ();
2434
2435         spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2436
2437         while (!ksocknal_data.ksnd_shuttingdown) {
2438
2439                 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2440                         route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2441                                             ksock_route_t, ksnr_connect_list);
2442                         
2443                         list_del (&route->ksnr_connect_list);
2444                         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2445
2446                         ksocknal_autoconnect (route);
2447                         ksocknal_put_route (route);
2448
2449                         spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2450                         continue;
2451                 }
2452                 
2453                 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2454
2455                 rc = wait_event_interruptible (ksocknal_data.ksnd_autoconnectd_waitq,
2456                                                ksocknal_data.ksnd_shuttingdown ||
2457                                                !list_empty (&ksocknal_data.ksnd_autoconnectd_routes));
2458
2459                 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2460         }
2461
2462         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2463
2464         ksocknal_thread_fini ();
2465         return (0);
2466 }
2467
2468 ksock_conn_t *
2469 ksocknal_find_timed_out_conn (ksock_peer_t *peer) 
2470 {
2471         /* We're called with a shared lock on ksnd_global_lock */
2472         ksock_conn_t      *conn;
2473         struct list_head  *ctmp;
2474         ksock_sched_t     *sched;
2475
2476         list_for_each (ctmp, &peer->ksnp_conns) {
2477                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2478                 sched = conn->ksnc_scheduler;
2479
2480                 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2481                 LASSERT (!conn->ksnc_closing);
2482                 
2483                 if (conn->ksnc_rx_started &&
2484                     time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2485                         /* Timed out incomplete incoming message */
2486                         atomic_inc (&conn->ksnc_refcount);
2487                         CERROR ("Timed out RX from "LPX64" %p %d.%d.%d.%d\n",
2488                                 peer->ksnp_nid, conn, HIPQUAD(conn->ksnc_ipaddr));
2489                         return (conn);
2490                 }
2491                 
2492                 if ((!list_empty (&conn->ksnc_tx_queue) ||
2493                      conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2494                     time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2495                         /* Timed out messages queued for sending, or
2496                          * messages buffered in the socket's send buffer */
2497                         atomic_inc (&conn->ksnc_refcount);
2498                         CERROR ("Timed out TX to "LPX64" %s%d %p %d.%d.%d.%d\n", 
2499                                 peer->ksnp_nid, 
2500                                 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2501                                 conn->ksnc_sock->sk->sk_wmem_queued, conn,
2502                                 HIPQUAD(conn->ksnc_ipaddr));
2503                         return (conn);
2504                 }
2505         }
2506
2507         return (NULL);
2508 }
2509
2510 void
2511 ksocknal_check_peer_timeouts (int idx)
2512 {
2513         struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2514         struct list_head *ptmp;
2515         ksock_peer_t     *peer;
2516         ksock_conn_t     *conn;
2517
2518  again:
2519         /* NB. We expect to have a look at all the peers and not find any
2520          * connections to time out, so we just use a shared lock while we
2521          * take a look... */
2522         read_lock (&ksocknal_data.ksnd_global_lock);
2523
2524         list_for_each (ptmp, peers) {
2525                 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2526                 conn = ksocknal_find_timed_out_conn (peer);
2527                 
2528                 if (conn != NULL) {
2529                         read_unlock (&ksocknal_data.ksnd_global_lock);
2530
2531                         CERROR ("Timeout out conn->"LPX64" ip %d.%d.%d.%d:%d\n",
2532                                 peer->ksnp_nid,
2533                                 HIPQUAD(conn->ksnc_ipaddr),
2534                                 conn->ksnc_port);
2535                         ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2536                         
2537                         /* NB we won't find this one again, but we can't
2538                          * just proceed with the next peer, since we dropped
2539                          * ksnd_global_lock and it might be dead already! */
2540                         ksocknal_put_conn (conn);
2541                         goto again;
2542                 }
2543         }
2544
2545         read_unlock (&ksocknal_data.ksnd_global_lock);
2546 }
2547
2548 int
2549 ksocknal_reaper (void *arg)
2550 {
2551         wait_queue_t       wait;
2552         unsigned long      flags;
2553         ksock_conn_t      *conn;
2554         ksock_sched_t     *sched;
2555         struct list_head   enomem_conns;
2556         int                nenomem_conns;
2557         int                timeout;
2558         int                i;
2559         int                peer_index = 0;
2560         unsigned long      deadline = jiffies;
2561         
2562         kportal_daemonize ("ksocknal_reaper");
2563         kportal_blockallsigs ();
2564
2565         INIT_LIST_HEAD(&enomem_conns);
2566         init_waitqueue_entry (&wait, current);
2567
2568         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2569
2570         while (!ksocknal_data.ksnd_shuttingdown) {
2571
2572                 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2573                         conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2574                                            ksock_conn_t, ksnc_list);
2575                         list_del (&conn->ksnc_list);
2576                         
2577                         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2578
2579                         ksocknal_terminate_conn (conn);
2580                         ksocknal_put_conn (conn);
2581
2582                         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2583                         continue;
2584                 }
2585
2586                 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2587                         conn = list_entry (ksocknal_data.ksnd_zombie_conns.next,
2588                                            ksock_conn_t, ksnc_list);
2589                         list_del (&conn->ksnc_list);
2590                         
2591                         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2592
2593                         ksocknal_destroy_conn (conn);
2594
2595                         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2596                         continue;
2597                 }
2598
2599                 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2600                         list_add(&enomem_conns, &ksocknal_data.ksnd_enomem_conns);
2601                         list_del_init(&ksocknal_data.ksnd_enomem_conns);
2602                 }
2603
2604                 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2605
2606                 /* reschedule all the connections that stalled with ENOMEM... */
2607                 nenomem_conns = 0;
2608                 while (!list_empty (&enomem_conns)) {
2609                         conn = list_entry (enomem_conns.next,
2610                                            ksock_conn_t, ksnc_tx_list);
2611                         list_del (&conn->ksnc_tx_list);
2612
2613                         sched = conn->ksnc_scheduler;
2614
2615                         spin_lock_irqsave (&sched->kss_lock, flags);
2616
2617                         LASSERT (conn->ksnc_tx_scheduled);
2618                         conn->ksnc_tx_ready = 1;
2619                         list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
2620                         wake_up (&sched->kss_waitq);
2621
2622                         spin_unlock_irqrestore (&sched->kss_lock, flags);
2623                         nenomem_conns++;
2624                 }
2625                 
2626                 /* careful with the jiffy wrap... */
2627                 while ((timeout = (int)(deadline - jiffies)) <= 0) {
2628                         const int n = 4;
2629                         const int p = 1;
2630                         int       chunk = ksocknal_data.ksnd_peer_hash_size;
2631                         
2632                         /* Time to check for timeouts on a few more peers: I do
2633                          * checks every 'p' seconds on a proportion of the peer
2634                          * table and I need to check every connection 'n' times
2635                          * within a timeout interval, to ensure I detect a
2636                          * timeout on any connection within (n+1)/n times the
2637                          * timeout interval. */
2638
2639                         if (ksocknal_data.ksnd_io_timeout > n * p)
2640                                 chunk = (chunk * n * p) / 
2641                                         ksocknal_data.ksnd_io_timeout;
2642                         if (chunk == 0)
2643                                 chunk = 1;
2644
2645                         for (i = 0; i < chunk; i++) {
2646                                 ksocknal_check_peer_timeouts (peer_index);
2647                                 peer_index = (peer_index + 1) % 
2648                                              ksocknal_data.ksnd_peer_hash_size;
2649                         }
2650
2651                         deadline += p * HZ;
2652                 }
2653
2654                 if (nenomem_conns != 0) {
2655                         /* Reduce my timeout if I rescheduled ENOMEM conns.
2656                          * This also prevents me getting woken immediately
2657                          * if any go back on my enomem list. */
2658                         timeout = SOCKNAL_ENOMEM_RETRY;
2659                 }
2660                 ksocknal_data.ksnd_reaper_waketime = jiffies + timeout;
2661
2662                 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2663                 set_current_state (TASK_INTERRUPTIBLE);
2664
2665                 if (!ksocknal_data.ksnd_shuttingdown &&
2666                     list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2667                     list_empty (&ksocknal_data.ksnd_zombie_conns))
2668                         schedule_timeout (timeout);
2669
2670                 set_current_state (TASK_RUNNING);
2671                 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2672
2673                 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2674         }
2675
2676         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2677
2678         ksocknal_thread_fini ();
2679         return (0);
2680 }
2681
2682 nal_cb_t ksocknal_lib = {
2683         nal_data:       &ksocknal_data,                /* NAL private data */
2684         cb_send:         ksocknal_send,
2685         cb_send_pages:   ksocknal_send_pages,
2686         cb_recv:         ksocknal_recv,
2687         cb_recv_pages:   ksocknal_recv_pages,
2688         cb_read:         ksocknal_read,
2689         cb_write:        ksocknal_write,
2690         cb_malloc:       ksocknal_malloc,
2691         cb_free:         ksocknal_free,
2692         cb_printf:       ksocknal_printf,
2693         cb_cli:          ksocknal_cli,
2694         cb_sti:          ksocknal_sti,
2695         cb_dist:         ksocknal_dist
2696 };