1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
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>
10 * This file is part of Portals, http://www.sf.net/projects/sandiaportals/
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.
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.
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.
29 * LIB functions follow
33 ksocknal_read(nal_cb_t *nal, void *private, void *dst_addr,
34 user_ptr src_addr, size_t len)
36 CDEBUG(D_NET, LPX64": reading %ld bytes from %p -> %p\n",
37 nal->ni.nid, (long)len, src_addr, dst_addr);
39 memcpy( dst_addr, src_addr, len );
44 ksocknal_write(nal_cb_t *nal, void *private, user_ptr dst_addr,
45 void *src_addr, size_t len)
47 CDEBUG(D_NET, LPX64": writing %ld bytes from %p -> %p\n",
48 nal->ni.nid, (long)len, src_addr, dst_addr);
50 memcpy( dst_addr, src_addr, len );
55 ksocknal_malloc(nal_cb_t *nal, size_t len)
59 PORTAL_ALLOC(buf, len);
68 ksocknal_free(nal_cb_t *nal, void *buf, size_t len)
70 PORTAL_FREE(buf, len);
74 ksocknal_printf(nal_cb_t *nal, const char *fmt, ...)
80 vsnprintf (msg, sizeof (msg), fmt, ap); /* sprint safely */
83 msg[sizeof (msg) - 1] = 0; /* ensure terminated */
85 CDEBUG (D_NET, "%s", msg);
89 ksocknal_cli(nal_cb_t *nal, unsigned long *flags)
91 ksock_nal_data_t *data = nal->nal_data;
93 spin_lock(&data->ksnd_nal_cb_lock);
97 ksocknal_sti(nal_cb_t *nal, unsigned long *flags)
99 ksock_nal_data_t *data;
100 data = nal->nal_data;
102 spin_unlock(&data->ksnd_nal_cb_lock);
106 ksocknal_dist(nal_cb_t *nal, ptl_nid_t nid, unsigned long *dist)
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 ) {
120 ksocknal_free_ltx (ksock_ltx_t *ltx)
122 atomic_dec(&ksocknal_data.ksnd_nactive_ltxs);
123 PORTAL_FREE(ltx, ltx->ltx_desc_size);
128 ksocknal_kvaddr_to_page (unsigned long vaddr)
132 if (vaddr >= VMALLOC_START &&
134 page = vmalloc_to_page ((void *)vaddr);
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 */
142 page = virt_to_page (vaddr);
153 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
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));
163 int offset = vaddr & (PAGE_SIZE - 1);
164 int zcsize = MIN (fragsize, PAGE_SIZE - offset);
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);
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) {
180 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
181 (void *)vaddr, page, page_address(page), offset, zcsize);
183 if (fragsize > zcsize) {
188 rc = tcp_sendpage_zccd(sock, page, offset, zcsize,
189 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
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 = {
204 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
206 mm_segment_t oldmm = get_fs();
209 rc = sock_sendmsg(sock, &msg, fragsize);
216 if (rc < iov->iov_len) {
217 /* didn't send whole iov entry... */
218 iov->iov_base = (void *)(vaddr + rc);
230 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
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));
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);
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))) {
253 CDEBUG(D_NET, "page %p + offset %x for %d\n",
254 page, offset, fragsize);
256 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
257 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
262 char *addr = ((char *)kmap (page)) + offset;
263 struct iovec fragiov = {.iov_base = addr,
264 .iov_len = fragsize};
265 struct msghdr msg = {
272 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
274 mm_segment_t oldmm = get_fs();
277 rc = sock_sendmsg(sock, &msg, fragsize);
287 kiov->kiov_offset = offset + rc;
288 kiov->kiov_len = fragsize - rc;
299 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
303 if (ksocknal_data.ksnd_stall_tx != 0) {
304 set_current_state (TASK_UNINTERRUPTIBLE);
305 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
308 LASSERT (tx->tx_resid != 0);
310 rc = ksocknal_getconnsock (conn);
312 LASSERT (conn->ksnc_closing);
317 if (ksocknal_data.ksnd_enomem_tx > 0) {
319 ksocknal_data.ksnd_enomem_tx--;
321 } else if (tx->tx_niov != 0) {
322 rc = ksocknal_send_iov (conn, tx);
324 rc = ksocknal_send_kiov (conn, tx);
328 /* Didn't write anything.
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).
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... */
340 LASSERT(rc != -ENOMEM);
342 if (rc == 0 || rc == -EAGAIN) {
343 if (test_bit(SOCK_NOSPACE,
344 &conn->ksnc_sock->flags)) {
350 if ((counter & (-counter)) == counter)
351 CWARN("%d ENOMEM tx %p\n",
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;
371 } while (tx->tx_resid != 0);
373 ksocknal_putconnsock (conn);
378 ksocknal_eager_ack (ksock_conn_t *conn)
381 mm_segment_t oldmm = get_fs();
382 struct socket *sock = conn->ksnc_sock;
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
390 sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
391 (char *)&opt, sizeof (opt));
396 ksocknal_recv_iov (ksock_conn_t *conn)
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 = {
412 mm_segment_t oldmm = get_fs();
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);
421 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
422 /* NB this is just a boolean............................^ */
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;
435 conn->ksnc_rx_nob_wanted -= rc;
436 conn->ksnc_rx_nob_left -= rc;
439 iov->iov_base = (void *)(vaddr + rc);
440 iov->iov_len = fragsize - rc;
445 conn->ksnc_rx_niov--;
450 ksocknal_recv_kiov (ksock_conn_t *conn)
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 = {
468 mm_segment_t oldmm = get_fs();
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);
478 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
479 /* NB this is just a boolean............................^ */
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;
494 conn->ksnc_rx_nob_wanted -= rc;
495 conn->ksnc_rx_nob_left -= rc;
498 kiov->kiov_offset = offset + rc;
499 kiov->kiov_len = fragsize - rc;
503 conn->ksnc_rx_kiov++;
504 conn->ksnc_rx_nkiov--;
509 ksocknal_receive (ksock_conn_t *conn)
511 /* Return 1 on success, 0 on EOF, < 0 on error.
512 * Caller checks ksnc_rx_nob_wanted to determine
513 * progress/completion. */
517 if (ksocknal_data.ksnd_stall_rx != 0) {
518 set_current_state (TASK_UNINTERRUPTIBLE);
519 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
522 rc = ksocknal_getconnsock (conn);
524 LASSERT (conn->ksnc_closing);
529 if (conn->ksnc_rx_niov != 0)
530 rc = ksocknal_recv_iov (conn);
532 rc = ksocknal_recv_kiov (conn);
535 /* error/EOF or partial receive */
538 } else if (rc == 0 && conn->ksnc_rx_started) {
539 /* EOF in the middle of a message */
545 /* Completed a fragment */
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);
560 ksocknal_putconnsock (conn);
566 ksocknal_zc_callback (zccd_t *zcd)
568 ksock_tx_t *tx = KSOCK_ZCCD_2_TX(zcd);
569 ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
573 /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
575 spin_lock_irqsave (&sched->kss_lock, flags);
577 list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
578 wake_up (&sched->kss_waitq);
580 spin_unlock_irqrestore (&sched->kss_lock, flags);
586 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
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);
595 /* zero copy completion isn't always from
596 * process_transmit() so it needs to keep a ref on
599 ksocknal_put_conn (tx->tx_conn);
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);
614 ltx = KSOCK_TX_2_KSOCK_LTX (tx);
616 lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie,
617 (tx->tx_resid == 0) ? PTL_OK : PTL_FAIL);
619 ksocknal_free_ltx (ltx);
624 ksocknal_tx_launched (ksock_tx_t *tx)
627 if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
628 ksock_conn_t *conn = tx->tx_conn;
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);
634 /* ...then drop the initial ref on zccd, so the zero copy
635 * callback can occur */
636 zccd_put (&tx->tx_zccd);
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);
646 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
651 rc = ksocknal_transmit (conn, tx);
653 CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
655 if (tx->tx_resid == 0) {
656 /* Sent everything OK */
659 ksocknal_tx_launched (tx);
667 /* Queue on ksnd_enomem_conns for retry after a timeout */
668 spin_lock_irqsave(&ksocknal_data.ksnd_reaper_lock, flags);
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);
678 spin_unlock_irqrestore(&ksocknal_data.ksnd_reaper_lock, flags);
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),
692 ksocknal_close_conn_and_siblings (conn, rc);
693 ksocknal_tx_launched (tx);
699 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
703 /* called holding write lock on ksnd_global_lock */
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);
710 if (ksocknal_data.ksnd_typed_conns)
711 route->ksnr_connecting =
712 KSNR_TYPED_ROUTES & ~route->ksnr_connected;
714 route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
716 atomic_inc (&route->ksnr_refcount); /* extra ref for asynchd */
718 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
720 list_add_tail (&route->ksnr_connect_list,
721 &ksocknal_data.ksnd_autoconnectd_routes);
722 wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
724 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
728 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
730 char ipbuf[PTL_NALFMT_SIZE];
731 ptl_nid_t target_nid;
733 ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
739 CERROR ("Can't send packet to "LPX64
740 " %s: routed target is not a peer\n",
741 nid, portals_nid2str(SOCKNAL, nid, ipbuf));
745 rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
748 CERROR ("Can't route to "LPX64" %s: router error %d\n",
749 nid, portals_nid2str(SOCKNAL, nid, ipbuf), rc);
753 peer = ksocknal_find_peer_locked (target_nid);
757 CERROR ("Can't send packet to "LPX64" %s: no peer entry\n",
758 target_nid, portals_nid2str(SOCKNAL, target_nid, ipbuf));
763 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
765 struct list_head *tmp;
766 ksock_conn_t *typed = NULL;
768 ksock_conn_t *fallback = NULL;
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);
776 LASSERT (!c->ksnc_closing);
778 if (fallback == NULL || nob < fnob) {
783 if (!ksocknal_data.ksnd_typed_conns)
786 switch (c->ksnc_type) {
789 case SOCKNAL_CONN_ANY:
791 case SOCKNAL_CONN_BULK_IN:
793 case SOCKNAL_CONN_BULK_OUT:
794 if (tx->tx_nob < ksocknal_data.ksnd_min_bulk)
797 case SOCKNAL_CONN_CONTROL:
798 if (tx->tx_nob >= ksocknal_data.ksnd_min_bulk)
803 if (typed == NULL || nob < tnob) {
809 /* prefer the typed selection */
810 return ((typed != NULL) ? typed : fallback);
814 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
817 ksock_sched_t *sched = conn->ksnc_scheduler;
819 /* called holding global lock (read or irq-write) and caller may
820 * not have dropped this lock between finding conn and calling me,
821 * so we don't need the {get,put}connsock dance to deref
823 LASSERT(!conn->ksnc_closing);
824 LASSERT(tx->tx_resid == tx->tx_nob);
826 CDEBUG (D_NET, "Sending to "LPX64" ip %d.%d.%d.%d:%d\n",
827 conn->ksnc_peer->ksnp_nid,
828 HIPQUAD(conn->ksnc_ipaddr),
831 atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
835 zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
836 /* NB this sets 1 ref on zccd, so the callback can only occur after
837 * I've released this ref. */
839 spin_lock_irqsave (&sched->kss_lock, flags);
841 conn->ksnc_tx_deadline = jiffies +
842 ksocknal_data.ksnd_io_timeout * HZ;
843 mb(); /* order with list_add_tail */
845 list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
847 if (conn->ksnc_tx_ready && /* able to send */
848 !conn->ksnc_tx_scheduled) { /* not scheduled to send */
849 /* +1 ref for scheduler */
850 atomic_inc (&conn->ksnc_refcount);
851 list_add_tail (&conn->ksnc_tx_list,
852 &sched->kss_tx_conns);
853 conn->ksnc_tx_scheduled = 1;
854 wake_up (&sched->kss_waitq);
857 spin_unlock_irqrestore (&sched->kss_lock, flags);
861 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
863 struct list_head *tmp;
864 ksock_route_t *route;
865 ksock_route_t *candidate = NULL;
869 list_for_each (tmp, &peer->ksnp_routes) {
870 route = list_entry (tmp, ksock_route_t, ksnr_list);
871 bits = route->ksnr_connected;
873 if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES ||
874 (bits & (1 << SOCKNAL_CONN_ANY)) != 0 ||
875 route->ksnr_connecting != 0) {
876 /* All typed connections have been established, or
877 * an untyped connection has been established, or
878 * connections are currently being established */
883 /* too soon to retry this guy? */
884 if (!time_after_eq (jiffies, route->ksnr_timeout))
887 /* always do eager routes */
888 if (route->ksnr_eager)
891 if (candidate == NULL) {
892 /* If we don't find any other route that is fully
893 * connected or connecting, the first connectable
894 * route is returned. If it fails to connect, it
895 * will get placed at the end of the list */
900 return (found ? NULL : candidate);
904 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
906 struct list_head *tmp;
907 ksock_route_t *route;
909 list_for_each (tmp, &peer->ksnp_routes) {
910 route = list_entry (tmp, ksock_route_t, ksnr_list);
912 if (route->ksnr_connecting != 0)
920 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
925 ksock_route_t *route;
928 /* Ensure the frags we've been given EXACTLY match the number of
929 * bytes we want to send. Many TCP/IP stacks disregard any total
930 * size parameters passed to them and just look at the frags.
932 * We always expect at least 1 mapped fragment containing the
933 * complete portals header. */
934 LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
935 lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
936 LASSERT (tx->tx_niov >= 1);
937 LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
939 CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
940 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type,
941 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
943 tx->tx_conn = NULL; /* only set when assigned a conn */
944 tx->tx_resid = tx->tx_nob;
945 tx->tx_hdr = (ptl_hdr_t *)tx->tx_iov[0].iov_base;
947 g_lock = &ksocknal_data.ksnd_global_lock;
950 peer = ksocknal_find_target_peer_locked (tx, nid);
952 read_unlock (g_lock);
953 return (-EHOSTUNREACH);
956 if (ksocknal_find_connectable_route_locked(peer) == NULL) {
957 conn = ksocknal_find_conn_locked (tx, peer);
959 /* I've got no autoconnect routes that need to be
960 * connecting and I do have an actual connection... */
961 ksocknal_queue_tx_locked (tx, conn);
962 read_unlock (g_lock);
967 /* Making one or more connections; I'll need a write lock... */
969 atomic_inc (&peer->ksnp_refcount); /* +1 ref for me while I unlock */
970 read_unlock (g_lock);
971 write_lock_irqsave (g_lock, flags);
973 if (peer->ksnp_closing) { /* peer deleted as I blocked! */
974 write_unlock_irqrestore (g_lock, flags);
975 ksocknal_put_peer (peer);
976 return (-EHOSTUNREACH);
978 ksocknal_put_peer (peer); /* drop ref I got above */
981 /* launch any/all autoconnections that need it */
982 route = ksocknal_find_connectable_route_locked (peer);
986 ksocknal_launch_autoconnect_locked (route);
989 conn = ksocknal_find_conn_locked (tx, peer);
991 /* Connection exists; queue message on it */
992 ksocknal_queue_tx_locked (tx, conn);
993 write_unlock_irqrestore (g_lock, flags);
997 route = ksocknal_find_connecting_route_locked (peer);
999 /* At least 1 connection is being established; queue the
1001 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
1002 write_unlock_irqrestore (g_lock, flags);
1006 write_unlock_irqrestore (g_lock, flags);
1007 return (-EHOSTUNREACH);
1011 ksocknal_sendmsg(nal_cb_t *nal,
1018 unsigned int payload_niov,
1019 struct iovec *payload_iov,
1020 ptl_kiov_t *payload_kiov,
1021 size_t payload_offset,
1028 /* NB 'private' is different depending on what we're sending.
1029 * Just ignore it... */
1031 CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid:"LPX64
1032 " pid %d\n", payload_nob, payload_niov, nid , pid);
1034 LASSERT (payload_nob == 0 || payload_niov > 0);
1035 LASSERT (payload_niov <= PTL_MD_MAX_IOV);
1037 /* It must be OK to kmap() if required */
1038 LASSERT (payload_kiov == NULL || !in_interrupt ());
1039 /* payload is either all vaddrs or all pages */
1040 LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
1042 if (payload_iov != NULL)
1043 desc_size = offsetof(ksock_ltx_t, ltx_iov[1 + payload_niov]);
1045 desc_size = offsetof(ksock_ltx_t, ltx_kiov[payload_niov]);
1047 if (in_interrupt() ||
1048 type == PTL_MSG_ACK ||
1049 type == PTL_MSG_REPLY) {
1050 /* Can't block if in interrupt or responding to an incoming
1052 PORTAL_ALLOC_ATOMIC(ltx, desc_size);
1054 PORTAL_ALLOC(ltx, desc_size);
1058 CERROR("Can't allocate tx desc type %d size %d %s\n",
1059 type, desc_size, in_interrupt() ? "(intr)" : "");
1060 return (PTL_NOSPACE);
1063 atomic_inc(&ksocknal_data.ksnd_nactive_ltxs);
1065 ltx->ltx_desc_size = desc_size;
1067 /* We always have 1 mapped frag for the header */
1068 ltx->ltx_tx.tx_iov = ltx->ltx_iov;
1069 ltx->ltx_iov[0].iov_base = <x->ltx_hdr;
1070 ltx->ltx_iov[0].iov_len = sizeof(*hdr);
1071 ltx->ltx_hdr = *hdr;
1073 ltx->ltx_private = private;
1074 ltx->ltx_cookie = cookie;
1076 ltx->ltx_tx.tx_isfwd = 0;
1077 ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_nob;
1079 if (payload_iov != NULL) {
1080 /* payload is all mapped */
1081 ltx->ltx_tx.tx_kiov = NULL;
1082 ltx->ltx_tx.tx_nkiov = 0;
1084 ltx->ltx_tx.tx_niov =
1085 1 + lib_extract_iov(payload_niov, <x->ltx_iov[1],
1086 payload_niov, payload_iov,
1087 payload_offset, payload_nob);
1089 /* payload is all pages */
1090 ltx->ltx_tx.tx_niov = 1;
1092 ltx->ltx_tx.tx_kiov = ltx->ltx_kiov;
1093 ltx->ltx_tx.tx_nkiov =
1094 lib_extract_kiov(payload_niov, ltx->ltx_kiov,
1095 payload_niov, payload_kiov,
1096 payload_offset, payload_nob);
1099 rc = ksocknal_launch_packet(<x->ltx_tx, nid);
1103 ksocknal_free_ltx(ltx);
1108 ksocknal_send (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1109 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1110 unsigned int payload_niov, struct iovec *payload_iov,
1111 size_t payload_offset, size_t payload_len)
1113 return (ksocknal_sendmsg(nal, private, cookie,
1114 hdr, type, nid, pid,
1115 payload_niov, payload_iov, NULL,
1116 payload_offset, payload_len));
1120 ksocknal_send_pages (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1121 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1122 unsigned int payload_niov, ptl_kiov_t *payload_kiov,
1123 size_t payload_offset, size_t payload_len)
1125 return (ksocknal_sendmsg(nal, private, cookie,
1126 hdr, type, nid, pid,
1127 payload_niov, NULL, payload_kiov,
1128 payload_offset, payload_len));
1132 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1134 ptl_nid_t nid = fwd->kprfd_gateway_nid;
1135 ksock_tx_t *tx = (ksock_tx_t *)&fwd->kprfd_scratch;
1138 CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1139 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1141 /* I'm the gateway; must be the last hop */
1142 if (nid == ksocknal_lib.ni.nid)
1143 nid = fwd->kprfd_target_nid;
1145 tx->tx_isfwd = 1; /* This is a forwarding packet */
1146 tx->tx_nob = fwd->kprfd_nob;
1147 tx->tx_niov = fwd->kprfd_niov;
1148 tx->tx_iov = fwd->kprfd_iov;
1152 rc = ksocknal_launch_packet (tx, nid);
1154 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
1158 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1160 long pid = kernel_thread (fn, arg, 0);
1165 atomic_inc (&ksocknal_data.ksnd_nthreads);
1170 ksocknal_thread_fini (void)
1172 atomic_dec (&ksocknal_data.ksnd_nthreads);
1176 ksocknal_fmb_callback (void *arg, int error)
1178 ksock_fmb_t *fmb = (ksock_fmb_t *)arg;
1179 ksock_fmb_pool_t *fmp = fmb->fmb_pool;
1180 ptl_hdr_t *hdr = (ptl_hdr_t *) page_address(fmb->fmb_pages[0]);
1181 ksock_conn_t *conn = NULL;
1182 ksock_sched_t *sched;
1183 unsigned long flags;
1184 char ipbuf[PTL_NALFMT_SIZE];
1185 char ipbuf2[PTL_NALFMT_SIZE];
1188 CERROR("Failed to route packet from "
1189 LPX64" %s to "LPX64" %s: %d\n",
1190 NTOH__u64(hdr->src_nid),
1191 portals_nid2str(SOCKNAL, NTOH__u64(hdr->src_nid), ipbuf),
1192 NTOH__u64(hdr->dest_nid),
1193 portals_nid2str(SOCKNAL, NTOH__u64(hdr->dest_nid), ipbuf2),
1196 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1197 NTOH__u64 (hdr->src_nid), NTOH__u64 (hdr->dest_nid));
1199 /* drop peer ref taken on init */
1200 ksocknal_put_peer (fmb->fmb_peer);
1202 spin_lock_irqsave (&fmp->fmp_lock, flags);
1204 list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1205 fmp->fmp_nactive_fmbs--;
1207 if (!list_empty (&fmp->fmp_blocked_conns)) {
1208 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1209 ksock_conn_t, ksnc_rx_list);
1210 list_del (&conn->ksnc_rx_list);
1213 spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1218 CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1219 LASSERT (conn->ksnc_rx_scheduled);
1220 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1222 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1224 sched = conn->ksnc_scheduler;
1226 spin_lock_irqsave (&sched->kss_lock, flags);
1228 list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1229 wake_up (&sched->kss_waitq);
1231 spin_unlock_irqrestore (&sched->kss_lock, flags);
1235 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1237 int payload_nob = conn->ksnc_rx_nob_left;
1238 int packet_nob = sizeof (ptl_hdr_t) + payload_nob;
1239 unsigned long flags;
1240 ksock_fmb_pool_t *pool;
1243 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1244 LASSERT (kpr_routing(&ksocknal_data.ksnd_router));
1246 if (packet_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1247 pool = &ksocknal_data.ksnd_small_fmp;
1249 pool = &ksocknal_data.ksnd_large_fmp;
1251 spin_lock_irqsave (&pool->fmp_lock, flags);
1253 if (!list_empty (&pool->fmp_idle_fmbs)) {
1254 fmb = list_entry(pool->fmp_idle_fmbs.next,
1255 ksock_fmb_t, fmb_list);
1256 list_del (&fmb->fmb_list);
1257 pool->fmp_nactive_fmbs++;
1258 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1263 /* deschedule until fmb free */
1265 conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1267 list_add_tail (&conn->ksnc_rx_list,
1268 &pool->fmp_blocked_conns);
1270 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1275 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1277 int payload_nob = conn->ksnc_rx_nob_left;
1278 int packet_nob = sizeof (ptl_hdr_t) + payload_nob;
1279 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1280 int niov; /* at least the header */
1283 LASSERT (conn->ksnc_rx_scheduled);
1284 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1285 LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1286 LASSERT (payload_nob >= 0);
1287 LASSERT (packet_nob <= fmb->fmb_npages * PAGE_SIZE);
1288 LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1290 /* Got a forwarding buffer; copy the header we just read into the
1291 * forwarding buffer. If there's payload, start reading reading it
1292 * into the buffer, otherwise the forwarding buffer can be kicked
1295 * NB fmb->fmb_iov spans the WHOLE packet.
1296 * conn->ksnc_rx_iov spans just the payload.
1298 fmb->fmb_iov[0].iov_base = page_address (fmb->fmb_pages[0]);
1301 memcpy (fmb->fmb_iov[0].iov_base, &conn->ksnc_hdr, sizeof (ptl_hdr_t));
1303 /* Take a ref on the conn's peer to prevent module unload before
1304 * forwarding completes. NB we ref peer and not conn since because
1305 * all refs on conn after it has been closed must remove themselves
1307 fmb->fmb_peer = conn->ksnc_peer;
1308 atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1310 if (payload_nob == 0) { /* got complete packet already */
1311 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (immediate)\n",
1312 conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1313 dest_nid, packet_nob);
1315 fmb->fmb_iov[0].iov_len = sizeof (ptl_hdr_t);
1317 kpr_fwd_init (&fmb->fmb_fwd, dest_nid,
1318 packet_nob, 1, fmb->fmb_iov,
1319 ksocknal_fmb_callback, fmb);
1321 /* forward it now */
1322 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1324 ksocknal_new_packet (conn, 0); /* on to next packet */
1329 if (packet_nob <= PAGE_SIZE) { /* whole packet fits in first page */
1330 fmb->fmb_iov[0].iov_len = packet_nob;
1332 fmb->fmb_iov[0].iov_len = PAGE_SIZE;
1333 nob = packet_nob - PAGE_SIZE;
1336 LASSERT (niov < fmb->fmb_npages);
1337 fmb->fmb_iov[niov].iov_base =
1338 page_address (fmb->fmb_pages[niov]);
1339 fmb->fmb_iov[niov].iov_len = MIN (PAGE_SIZE, nob);
1345 kpr_fwd_init (&fmb->fmb_fwd, dest_nid,
1346 packet_nob, niov, fmb->fmb_iov,
1347 ksocknal_fmb_callback, fmb);
1349 conn->ksnc_cookie = fmb; /* stash fmb for later */
1350 conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1352 /* payload is desc's iov-ed buffer, but skipping the hdr */
1353 LASSERT (niov <= sizeof (conn->ksnc_rx_iov_space) /
1354 sizeof (struct iovec));
1356 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1357 conn->ksnc_rx_iov[0].iov_base =
1358 (void *)(((unsigned long)fmb->fmb_iov[0].iov_base) +
1359 sizeof (ptl_hdr_t));
1360 conn->ksnc_rx_iov[0].iov_len =
1361 fmb->fmb_iov[0].iov_len - sizeof (ptl_hdr_t);
1364 memcpy(&conn->ksnc_rx_iov[1], &fmb->fmb_iov[1],
1365 (niov - 1) * sizeof (struct iovec));
1367 conn->ksnc_rx_niov = niov;
1369 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1370 NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1375 ksocknal_fwd_parse (ksock_conn_t *conn)
1378 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1379 ptl_nid_t src_nid = NTOH__u64 (conn->ksnc_hdr.src_nid);
1380 int body_len = NTOH__u32 (conn->ksnc_hdr.payload_length);
1381 char str[PTL_NALFMT_SIZE];
1382 char str2[PTL_NALFMT_SIZE];
1384 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1385 src_nid, dest_nid, conn->ksnc_rx_nob_left);
1387 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1388 LASSERT (conn->ksnc_rx_scheduled);
1390 if (body_len < 0) { /* length corrupt (overflow) */
1391 CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
1392 "packet size %d illegal\n",
1393 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1394 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1397 ksocknal_new_packet (conn, 0); /* on to new packet */
1401 if (!kpr_routing(&ksocknal_data.ksnd_router)) { /* not forwarding */
1402 CERROR("dropping packet from "LPX64" (%s) for "LPX64
1403 " (%s): not forwarding\n",
1404 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1405 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1406 /* on to new packet (skip this one's body) */
1407 ksocknal_new_packet (conn, body_len);
1411 if (body_len > PTL_MTU) { /* too big to forward */
1412 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1413 "(%s): packet size %d too big\n",
1414 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1415 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1417 /* on to new packet (skip this one's body) */
1418 ksocknal_new_packet (conn, body_len);
1422 /* should have gone direct */
1423 peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1425 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1426 "(%s): target is a peer\n",
1427 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1428 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1429 ksocknal_put_peer (peer); /* drop ref from get above */
1431 /* on to next packet (skip this one's body) */
1432 ksocknal_new_packet (conn, body_len);
1436 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB; /* Getting FMB now */
1437 conn->ksnc_rx_nob_left = body_len; /* stash packet size */
1438 conn->ksnc_rx_nob_wanted = body_len; /* (no slop) */
1442 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1444 static char ksocknal_slop_buffer[4096];
1450 if (nob_to_skip == 0) { /* right at next packet boundary now */
1451 conn->ksnc_rx_started = 0;
1452 mb (); /* racing with timeout thread */
1454 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1455 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1456 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1458 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1459 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1460 conn->ksnc_rx_iov[0].iov_len = sizeof (ptl_hdr_t);
1461 conn->ksnc_rx_niov = 1;
1463 conn->ksnc_rx_kiov = NULL;
1464 conn->ksnc_rx_nkiov = 0;
1468 /* Set up to skip as much a possible now. If there's more left
1469 * (ran out of iov entries) we'll get called again */
1471 conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1472 conn->ksnc_rx_nob_left = nob_to_skip;
1473 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1478 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1480 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1481 conn->ksnc_rx_iov[niov].iov_len = nob;
1486 } while (nob_to_skip != 0 && /* mustn't overflow conn's rx iov */
1487 niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1489 conn->ksnc_rx_niov = niov;
1490 conn->ksnc_rx_kiov = NULL;
1491 conn->ksnc_rx_nkiov = 0;
1492 conn->ksnc_rx_nob_wanted = skipped;
1497 ksocknal_process_receive (ksock_conn_t *conn)
1502 LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1504 /* doesn't need a forwarding buffer */
1505 if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1509 fmb = ksocknal_get_idle_fmb (conn);
1511 /* conn descheduled waiting for idle fmb */
1515 if (ksocknal_init_fmb (conn, fmb)) {
1516 /* packet forwarded */
1521 /* NB: sched lock NOT held */
1522 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1523 conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1524 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1525 conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1527 LASSERT (conn->ksnc_rx_nob_wanted > 0);
1529 rc = ksocknal_receive(conn);
1532 LASSERT (rc != -EAGAIN);
1535 CWARN ("[%p] EOF from "LPX64" ip %d.%d.%d.%d:%d\n",
1536 conn, conn->ksnc_peer->ksnp_nid,
1537 HIPQUAD(conn->ksnc_ipaddr),
1539 else if (!conn->ksnc_closing)
1540 CERROR ("[%p] Error %d on read from "LPX64
1541 " ip %d.%d.%d.%d:%d\n",
1542 conn, rc, conn->ksnc_peer->ksnp_nid,
1543 HIPQUAD(conn->ksnc_ipaddr),
1546 ksocknal_close_conn_and_siblings (conn, rc);
1547 return (rc == 0 ? -ESHUTDOWN : rc);
1550 if (conn->ksnc_rx_nob_wanted != 0) {
1555 switch (conn->ksnc_rx_state) {
1556 case SOCKNAL_RX_HEADER:
1557 if (conn->ksnc_hdr.type != HTON__u32(PTL_MSG_HELLO) &&
1558 NTOH__u64(conn->ksnc_hdr.dest_nid) != ksocknal_lib.ni.nid) {
1559 /* This packet isn't for me */
1560 ksocknal_fwd_parse (conn);
1561 switch (conn->ksnc_rx_state) {
1562 case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1563 return (0); /* => come back later */
1564 case SOCKNAL_RX_SLOP: /* skipping packet's body */
1565 goto try_read; /* => go read it */
1566 case SOCKNAL_RX_GET_FMB: /* forwarding */
1567 goto get_fmb; /* => go get a fwd msg buffer */
1574 /* sets wanted_len, iovs etc */
1575 lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1577 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1578 conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1579 goto try_read; /* go read the payload */
1581 /* Fall through (completed packet for me) */
1583 case SOCKNAL_RX_BODY:
1584 /* payload all received */
1585 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_OK);
1588 case SOCKNAL_RX_SLOP:
1589 /* starting new packet? */
1590 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1591 return (0); /* come back later */
1592 goto try_read; /* try to finish reading slop now */
1594 case SOCKNAL_RX_BODY_FWD:
1595 /* payload all received */
1596 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1597 conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1598 NTOH__u64 (conn->ksnc_hdr.dest_nid),
1599 conn->ksnc_rx_nob_left);
1601 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1602 * conn->ksnc_cookie */
1603 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1604 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1606 /* no slop in forwarded packets */
1607 LASSERT (conn->ksnc_rx_nob_left == 0);
1609 ksocknal_new_packet (conn, 0); /* on to next packet */
1610 return (0); /* (later) */
1618 return (-EINVAL); /* keep gcc happy */
1622 ksocknal_recv (nal_cb_t *nal, void *private, lib_msg_t *msg,
1623 unsigned int niov, struct iovec *iov,
1624 size_t offset, size_t mlen, size_t rlen)
1626 ksock_conn_t *conn = (ksock_conn_t *)private;
1628 LASSERT (mlen <= rlen);
1629 LASSERT (niov <= PTL_MD_MAX_IOV);
1631 conn->ksnc_cookie = msg;
1632 conn->ksnc_rx_nob_wanted = mlen;
1633 conn->ksnc_rx_nob_left = rlen;
1635 conn->ksnc_rx_nkiov = 0;
1636 conn->ksnc_rx_kiov = NULL;
1637 conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1638 conn->ksnc_rx_niov =
1639 lib_extract_iov(PTL_MD_MAX_IOV, conn->ksnc_rx_iov,
1640 niov, iov, offset, mlen);
1643 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1644 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1650 ksocknal_recv_pages (nal_cb_t *nal, void *private, lib_msg_t *msg,
1651 unsigned int niov, ptl_kiov_t *kiov,
1652 size_t offset, size_t mlen, size_t rlen)
1654 ksock_conn_t *conn = (ksock_conn_t *)private;
1656 LASSERT (mlen <= rlen);
1657 LASSERT (niov <= PTL_MD_MAX_IOV);
1659 conn->ksnc_cookie = msg;
1660 conn->ksnc_rx_nob_wanted = mlen;
1661 conn->ksnc_rx_nob_left = rlen;
1663 conn->ksnc_rx_niov = 0;
1664 conn->ksnc_rx_iov = NULL;
1665 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1666 conn->ksnc_rx_nkiov =
1667 lib_extract_kiov(PTL_MD_MAX_IOV, conn->ksnc_rx_kiov,
1668 niov, kiov, offset, mlen);
1671 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1672 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1677 int ksocknal_scheduler (void *arg)
1679 ksock_sched_t *sched = (ksock_sched_t *)arg;
1682 unsigned long flags;
1685 int id = sched - ksocknal_data.ksnd_schedulers;
1688 snprintf (name, sizeof (name),"ksocknald_%02d", id);
1689 kportal_daemonize (name);
1690 kportal_blockallsigs ();
1692 #if (CONFIG_SMP && CPU_AFFINITY)
1693 if ((cpu_online_map & (1 << id)) != 0) {
1695 current->cpus_allowed = (1 << id);
1697 set_cpus_allowed (current, 1<<id);
1700 CERROR ("Can't set CPU affinity for %s\n", name);
1702 #endif /* CONFIG_SMP && CPU_AFFINITY */
1704 spin_lock_irqsave (&sched->kss_lock, flags);
1706 while (!ksocknal_data.ksnd_shuttingdown) {
1707 int did_something = 0;
1709 /* Ensure I progress everything semi-fairly */
1711 if (!list_empty (&sched->kss_rx_conns)) {
1712 conn = list_entry(sched->kss_rx_conns.next,
1713 ksock_conn_t, ksnc_rx_list);
1714 list_del(&conn->ksnc_rx_list);
1716 LASSERT(conn->ksnc_rx_scheduled);
1717 LASSERT(conn->ksnc_rx_ready);
1719 /* clear rx_ready in case receive isn't complete.
1720 * Do it BEFORE we call process_recv, since
1721 * data_ready can set it any time after we release
1723 conn->ksnc_rx_ready = 0;
1724 spin_unlock_irqrestore(&sched->kss_lock, flags);
1726 rc = ksocknal_process_receive(conn);
1728 spin_lock_irqsave(&sched->kss_lock, flags);
1730 /* I'm the only one that can clear this flag */
1731 LASSERT(conn->ksnc_rx_scheduled);
1733 /* Did process_receive get everything it wanted? */
1735 conn->ksnc_rx_ready = 1;
1737 if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
1738 conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
1739 /* Conn blocked for a forwarding buffer.
1740 * It will get queued for my attention when
1741 * one becomes available (and it might just
1742 * already have been!). Meanwhile my ref
1743 * on it stays put. */
1744 } else if (conn->ksnc_rx_ready) {
1745 /* reschedule for rx */
1746 list_add_tail (&conn->ksnc_rx_list,
1747 &sched->kss_rx_conns);
1749 conn->ksnc_rx_scheduled = 0;
1751 ksocknal_put_conn(conn);
1757 if (!list_empty (&sched->kss_tx_conns)) {
1758 conn = list_entry(sched->kss_tx_conns.next,
1759 ksock_conn_t, ksnc_tx_list);
1760 list_del (&conn->ksnc_tx_list);
1762 LASSERT(conn->ksnc_tx_scheduled);
1763 LASSERT(conn->ksnc_tx_ready);
1764 LASSERT(!list_empty(&conn->ksnc_tx_queue));
1766 tx = list_entry(conn->ksnc_tx_queue.next,
1767 ksock_tx_t, tx_list);
1768 /* dequeue now so empty list => more to send */
1769 list_del(&tx->tx_list);
1771 /* Clear tx_ready in case send isn't complete. Do
1772 * it BEFORE we call process_transmit, since
1773 * write_space can set it any time after we release
1775 conn->ksnc_tx_ready = 0;
1776 spin_unlock_irqrestore (&sched->kss_lock, flags);
1778 rc = ksocknal_process_transmit(conn, tx);
1780 spin_lock_irqsave (&sched->kss_lock, flags);
1782 if (rc == -ENOMEM || rc == -EAGAIN) {
1783 /* Incomplete send: replace tx on HEAD of tx_queue */
1784 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
1786 /* Complete send; assume space for more */
1787 conn->ksnc_tx_ready = 1;
1790 if (rc == -ENOMEM) {
1791 /* Do nothing; after a short timeout, this
1792 * conn will be reposted on kss_tx_conns. */
1793 } else if (conn->ksnc_tx_ready &&
1794 !list_empty (&conn->ksnc_tx_queue)) {
1795 /* reschedule for tx */
1796 list_add_tail (&conn->ksnc_tx_list,
1797 &sched->kss_tx_conns);
1799 conn->ksnc_tx_scheduled = 0;
1801 ksocknal_put_conn (conn);
1807 if (!list_empty (&sched->kss_zctxdone_list)) {
1809 list_entry(sched->kss_zctxdone_list.next,
1810 ksock_tx_t, tx_list);
1813 list_del (&tx->tx_list);
1814 spin_unlock_irqrestore (&sched->kss_lock, flags);
1816 ksocknal_tx_done (tx, 1);
1818 spin_lock_irqsave (&sched->kss_lock, flags);
1821 if (!did_something || /* nothing to do */
1822 ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1823 spin_unlock_irqrestore (&sched->kss_lock, flags);
1827 if (!did_something) { /* wait for something to do */
1829 rc = wait_event_interruptible (sched->kss_waitq,
1830 ksocknal_data.ksnd_shuttingdown ||
1831 !list_empty(&sched->kss_rx_conns) ||
1832 !list_empty(&sched->kss_tx_conns) ||
1833 !list_empty(&sched->kss_zctxdone_list));
1835 rc = wait_event_interruptible (sched->kss_waitq,
1836 ksocknal_data.ksnd_shuttingdown ||
1837 !list_empty(&sched->kss_rx_conns) ||
1838 !list_empty(&sched->kss_tx_conns));
1844 spin_lock_irqsave (&sched->kss_lock, flags);
1848 spin_unlock_irqrestore (&sched->kss_lock, flags);
1849 ksocknal_thread_fini ();
1854 ksocknal_data_ready (struct sock *sk, int n)
1856 unsigned long flags;
1858 ksock_sched_t *sched;
1861 /* interleave correctly with closing sockets... */
1862 read_lock (&ksocknal_data.ksnd_global_lock);
1864 conn = sk->sk_user_data;
1865 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1866 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1867 sk->sk_data_ready (sk, n);
1869 sched = conn->ksnc_scheduler;
1871 spin_lock_irqsave (&sched->kss_lock, flags);
1873 conn->ksnc_rx_ready = 1;
1875 if (!conn->ksnc_rx_scheduled) { /* not being progressed */
1876 list_add_tail(&conn->ksnc_rx_list,
1877 &sched->kss_rx_conns);
1878 conn->ksnc_rx_scheduled = 1;
1879 /* extra ref for scheduler */
1880 atomic_inc (&conn->ksnc_refcount);
1882 wake_up (&sched->kss_waitq);
1885 spin_unlock_irqrestore (&sched->kss_lock, flags);
1888 read_unlock (&ksocknal_data.ksnd_global_lock);
1894 ksocknal_write_space (struct sock *sk)
1896 unsigned long flags;
1898 ksock_sched_t *sched;
1900 /* interleave correctly with closing sockets... */
1901 read_lock (&ksocknal_data.ksnd_global_lock);
1903 conn = sk->sk_user_data;
1905 CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1906 sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1907 (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1908 " ready" : " blocked"),
1909 (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1910 " scheduled" : " idle"),
1911 (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1912 " empty" : " queued"));
1914 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1915 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1916 sk->sk_write_space (sk);
1918 read_unlock (&ksocknal_data.ksnd_global_lock);
1922 if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1923 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1925 sched = conn->ksnc_scheduler;
1927 spin_lock_irqsave (&sched->kss_lock, flags);
1929 conn->ksnc_tx_ready = 1;
1931 if (!conn->ksnc_tx_scheduled && // not being progressed
1932 !list_empty(&conn->ksnc_tx_queue)){//packets to send
1933 list_add_tail (&conn->ksnc_tx_list,
1934 &sched->kss_tx_conns);
1935 conn->ksnc_tx_scheduled = 1;
1936 /* extra ref for scheduler */
1937 atomic_inc (&conn->ksnc_refcount);
1939 wake_up (&sched->kss_waitq);
1942 spin_unlock_irqrestore (&sched->kss_lock, flags);
1945 read_unlock (&ksocknal_data.ksnd_global_lock);
1949 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1952 mm_segment_t oldmm = get_fs();
1955 struct iovec iov = {
1959 struct msghdr msg = {
1964 .msg_control = NULL,
1965 .msg_controllen = 0,
1970 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1977 CERROR ("Unexpected zero rc\n");
1978 return (-ECONNABORTED);
1981 buffer = ((char *)buffer) + rc;
1989 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1992 mm_segment_t oldmm = get_fs();
1995 struct iovec iov = {
1999 struct msghdr msg = {
2004 .msg_control = NULL,
2005 .msg_controllen = 0,
2010 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
2017 return (-ECONNABORTED);
2019 buffer = ((char *)buffer) + rc;
2027 ksocknal_hello (struct socket *sock, ptl_nid_t *nid, int *type,
2032 ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
2033 char ipbuf[PTL_NALFMT_SIZE];
2034 char ipbuf2[PTL_NALFMT_SIZE];
2036 LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
2038 memset (&hdr, 0, sizeof (hdr));
2039 hmv->magic = __cpu_to_le32 (PORTALS_PROTO_MAGIC);
2040 hmv->version_major = __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR);
2041 hmv->version_minor = __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR);
2043 hdr.src_nid = __cpu_to_le64 (ksocknal_lib.ni.nid);
2044 hdr.type = __cpu_to_le32 (PTL_MSG_HELLO);
2046 hdr.msg.hello.type = __cpu_to_le32 (*type);
2047 hdr.msg.hello.incarnation =
2048 __cpu_to_le64 (ksocknal_data.ksnd_incarnation);
2050 /* Assume sufficient socket buffering for this message */
2051 rc = ksocknal_sock_write (sock, &hdr, sizeof (hdr));
2053 CERROR ("Error %d sending HELLO to "LPX64" %s\n",
2054 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2058 rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
2060 CERROR ("Error %d reading HELLO from "LPX64" %s\n",
2061 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2065 if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
2066 CERROR ("Bad magic %#08x (%#08x expected) from "LPX64" %s\n",
2067 __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, *nid,
2068 portals_nid2str(SOCKNAL, *nid, ipbuf));
2072 if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
2073 hmv->version_minor != __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
2074 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
2075 " from "LPX64" %s\n",
2076 __le16_to_cpu (hmv->version_major),
2077 __le16_to_cpu (hmv->version_minor),
2078 PORTALS_PROTO_VERSION_MAJOR,
2079 PORTALS_PROTO_VERSION_MINOR,
2080 *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2084 #if (PORTALS_PROTO_VERSION_MAJOR != 0)
2085 # error "This code only understands protocol version 0.x"
2087 /* version 0 sends magic/version as the dest_nid of a 'hello' header,
2088 * so read the rest of it in now... */
2090 rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
2092 CERROR ("Error %d reading rest of HELLO hdr from "LPX64" %s\n",
2093 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2097 /* ...and check we got what we expected */
2098 if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
2099 hdr.payload_length != __cpu_to_le32 (0)) {
2100 CERROR ("Expecting a HELLO hdr with 0 payload,"
2101 " but got type %d with %d payload from "LPX64" %s\n",
2102 __le32_to_cpu (hdr.type),
2103 __le32_to_cpu (hdr.payload_length), *nid,
2104 portals_nid2str(SOCKNAL, *nid, ipbuf));
2108 if (__le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
2109 CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY\n");
2113 if (*nid == PTL_NID_ANY) { /* don't know peer's nid yet */
2114 *nid = __le64_to_cpu(hdr.src_nid);
2115 } else if (*nid != __le64_to_cpu (hdr.src_nid)) {
2116 CERROR ("Connected to nid "LPX64" %s, but expecting "LPX64" %s\n",
2117 __le64_to_cpu (hdr.src_nid),
2118 portals_nid2str(SOCKNAL,
2119 __le64_to_cpu(hdr.src_nid),
2121 *nid, portals_nid2str(SOCKNAL, *nid, ipbuf2));
2125 if (*type == SOCKNAL_CONN_NONE) {
2126 /* I've accepted this connection; peer determines type */
2127 *type = __le32_to_cpu(hdr.msg.hello.type);
2129 case SOCKNAL_CONN_ANY:
2130 case SOCKNAL_CONN_CONTROL:
2132 case SOCKNAL_CONN_BULK_IN:
2133 *type = SOCKNAL_CONN_BULK_OUT;
2135 case SOCKNAL_CONN_BULK_OUT:
2136 *type = SOCKNAL_CONN_BULK_IN;
2139 CERROR ("Unexpected type %d from "LPX64" %s\n",
2141 portals_nid2str(SOCKNAL, *nid, ipbuf));
2144 } else if (__le32_to_cpu(hdr.msg.hello.type) != SOCKNAL_CONN_NONE) {
2145 CERROR ("Mismatched types: me %d "LPX64" %s %d\n",
2146 *type, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf),
2147 __le32_to_cpu(hdr.msg.hello.type));
2151 *incarnation = __le64_to_cpu(hdr.msg.hello.incarnation);
2157 ksocknal_setup_sock (struct socket *sock)
2159 mm_segment_t oldmm = get_fs ();
2162 struct linger linger;
2164 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
2165 sock->sk->sk_allocation = GFP_NOFS;
2167 sock->sk->allocation = GFP_NOFS;
2170 /* Ensure this socket aborts active sends immediately when we close
2174 linger.l_linger = 0;
2177 rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2178 (char *)&linger, sizeof (linger));
2181 CERROR ("Can't set SO_LINGER: %d\n", rc);
2187 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2188 (char *)&option, sizeof (option));
2191 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2195 #if SOCKNAL_USE_KEEPALIVES
2196 /* Keepalives: If 3/4 of the timeout elapses, start probing every
2197 * second until the timeout elapses. */
2199 option = (ksocknal_data.ksnd_io_timeout * 3) / 4;
2201 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2202 (char *)&option, sizeof (option));
2205 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2211 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2212 (char *)&option, sizeof (option));
2215 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2219 option = ksocknal_data.ksnd_io_timeout / 4;
2221 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2222 (char *)&option, sizeof (option));
2225 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2231 rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
2232 (char *)&option, sizeof (option));
2235 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2243 ksocknal_connect_peer (ksock_route_t *route, int type)
2245 struct sockaddr_in peer_addr;
2246 mm_segment_t oldmm = get_fs();
2249 struct socket *sock;
2251 char ipbuf[PTL_NALFMT_SIZE];
2253 rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2255 CERROR ("Can't create autoconnect socket: %d\n", rc);
2259 /* Ugh; have to map_fd for compatibility with sockets passed in
2260 * from userspace. And we actually need the sock->file refcounting
2261 * that this gives you :) */
2263 fd = sock_map_fd (sock);
2265 sock_release (sock);
2266 CERROR ("sock_map_fd error %d\n", fd);
2270 /* NB the fd now owns the ref on sock->file */
2271 LASSERT (sock->file != NULL);
2272 LASSERT (file_count(sock->file) == 1);
2274 /* Set the socket timeouts, so our connection attempt completes in
2276 tv.tv_sec = ksocknal_data.ksnd_io_timeout;
2280 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2281 (char *)&tv, sizeof (tv));
2284 CERROR ("Can't set send timeout %d: %d\n",
2285 ksocknal_data.ksnd_io_timeout, rc);
2290 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2291 (char *)&tv, sizeof (tv));
2294 CERROR ("Can't set receive timeout %d: %d\n",
2295 ksocknal_data.ksnd_io_timeout, rc);
2303 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2304 (char *)&option, sizeof (option));
2307 CERROR ("Can't disable nagle: %d\n", rc);
2312 if (route->ksnr_buffer_size != 0) {
2313 int option = route->ksnr_buffer_size;
2316 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2317 (char *)&option, sizeof (option));
2320 CERROR ("Can't set send buffer %d: %d\n",
2321 route->ksnr_buffer_size, rc);
2326 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2327 (char *)&option, sizeof (option));
2330 CERROR ("Can't set receive buffer %d: %d\n",
2331 route->ksnr_buffer_size, rc);
2336 memset (&peer_addr, 0, sizeof (peer_addr));
2337 peer_addr.sin_family = AF_INET;
2338 peer_addr.sin_port = htons (route->ksnr_port);
2339 peer_addr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2341 rc = sock->ops->connect (sock, (struct sockaddr *)&peer_addr,
2342 sizeof (peer_addr), sock->file->f_flags);
2344 CERROR ("Error %d connecting to "LPX64" %s\n", rc,
2345 route->ksnr_peer->ksnp_nid,
2346 portals_nid2str(SOCKNAL,
2347 route->ksnr_peer->ksnp_nid,
2352 rc = ksocknal_create_conn (route, sock, route->ksnr_irq_affinity, type);
2354 /* Take an extra ref on sock->file to compensate for the
2355 * upcoming close which will lose fd's ref on it. */
2356 get_file (sock->file);
2365 ksocknal_autoconnect (ksock_route_t *route)
2367 LIST_HEAD (zombies);
2370 unsigned long flags;
2375 for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
2376 if ((route->ksnr_connecting & (1 << type)) != 0)
2378 LASSERT (type < SOCKNAL_CONN_NTYPES);
2380 rc = ksocknal_connect_peer (route, type);
2385 /* successfully autoconnected: create_conn did the
2386 * route/conn binding and scheduled any blocked packets */
2388 if (route->ksnr_connecting == 0) {
2389 /* No more connections required */
2394 /* Connection attempt failed */
2396 write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2398 peer = route->ksnr_peer;
2399 route->ksnr_connecting = 0;
2401 /* This is a retry rather than a new connection */
2402 LASSERT (route->ksnr_retry_interval != 0);
2403 route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2404 route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2405 SOCKNAL_MAX_RECONNECT_INTERVAL);
2407 if (!list_empty (&peer->ksnp_tx_queue) &&
2408 ksocknal_find_connecting_route_locked (peer) == NULL) {
2409 LASSERT (list_empty (&peer->ksnp_conns));
2411 /* None of the connections that the blocked packets are
2412 * waiting for have been successful. Complete them now... */
2414 tx = list_entry (peer->ksnp_tx_queue.next,
2415 ksock_tx_t, tx_list);
2416 list_del (&tx->tx_list);
2417 list_add_tail (&tx->tx_list, &zombies);
2418 } while (!list_empty (&peer->ksnp_tx_queue));
2421 /* make this route least-favourite for re-selection */
2422 if (!route->ksnr_deleted) {
2423 list_del(&route->ksnr_list);
2424 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
2427 write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2429 while (!list_empty (&zombies)) {
2430 char ipbuf[PTL_NALFMT_SIZE];
2431 char ipbuf2[PTL_NALFMT_SIZE];
2432 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2434 CERROR ("Deleting packet type %d len %d ("LPX64" %s->"LPX64" %s)\n",
2435 NTOH__u32 (tx->tx_hdr->type),
2436 NTOH__u32 (tx->tx_hdr->payload_length),
2437 NTOH__u64 (tx->tx_hdr->src_nid),
2438 portals_nid2str(SOCKNAL,
2439 NTOH__u64(tx->tx_hdr->src_nid),
2441 NTOH__u64 (tx->tx_hdr->dest_nid),
2442 portals_nid2str(SOCKNAL,
2443 NTOH__u64(tx->tx_hdr->src_nid),
2446 list_del (&tx->tx_list);
2448 ksocknal_tx_done (tx, 0);
2453 ksocknal_autoconnectd (void *arg)
2455 long id = (long)arg;
2457 unsigned long flags;
2458 ksock_route_t *route;
2461 snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2462 kportal_daemonize (name);
2463 kportal_blockallsigs ();
2465 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2467 while (!ksocknal_data.ksnd_shuttingdown) {
2469 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2470 route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2471 ksock_route_t, ksnr_connect_list);
2473 list_del (&route->ksnr_connect_list);
2474 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2476 ksocknal_autoconnect (route);
2477 ksocknal_put_route (route);
2479 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2483 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2485 rc = wait_event_interruptible (ksocknal_data.ksnd_autoconnectd_waitq,
2486 ksocknal_data.ksnd_shuttingdown ||
2487 !list_empty (&ksocknal_data.ksnd_autoconnectd_routes));
2489 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2492 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2494 ksocknal_thread_fini ();
2499 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2501 /* We're called with a shared lock on ksnd_global_lock */
2503 struct list_head *ctmp;
2504 ksock_sched_t *sched;
2506 list_for_each (ctmp, &peer->ksnp_conns) {
2507 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2508 sched = conn->ksnc_scheduler;
2510 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2511 LASSERT (!conn->ksnc_closing);
2513 if (conn->ksnc_rx_started &&
2514 time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2515 /* Timed out incomplete incoming message */
2516 atomic_inc (&conn->ksnc_refcount);
2517 CERROR ("Timed out RX from "LPX64" %p %d.%d.%d.%d\n",
2518 peer->ksnp_nid, conn, HIPQUAD(conn->ksnc_ipaddr));
2522 if ((!list_empty (&conn->ksnc_tx_queue) ||
2523 conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2524 time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2525 /* Timed out messages queued for sending, or
2526 * messages buffered in the socket's send buffer */
2527 atomic_inc (&conn->ksnc_refcount);
2528 CERROR ("Timed out TX to "LPX64" %s%d %p %d.%d.%d.%d\n",
2530 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2531 conn->ksnc_sock->sk->sk_wmem_queued, conn,
2532 HIPQUAD(conn->ksnc_ipaddr));
2541 ksocknal_check_peer_timeouts (int idx)
2543 struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2544 struct list_head *ptmp;
2549 /* NB. We expect to have a look at all the peers and not find any
2550 * connections to time out, so we just use a shared lock while we
2552 read_lock (&ksocknal_data.ksnd_global_lock);
2554 list_for_each (ptmp, peers) {
2555 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2556 conn = ksocknal_find_timed_out_conn (peer);
2559 read_unlock (&ksocknal_data.ksnd_global_lock);
2561 CERROR ("Timeout out conn->"LPX64" ip %d.%d.%d.%d:%d\n",
2563 HIPQUAD(conn->ksnc_ipaddr),
2565 ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2567 /* NB we won't find this one again, but we can't
2568 * just proceed with the next peer, since we dropped
2569 * ksnd_global_lock and it might be dead already! */
2570 ksocknal_put_conn (conn);
2575 read_unlock (&ksocknal_data.ksnd_global_lock);
2579 ksocknal_reaper (void *arg)
2582 unsigned long flags;
2584 ksock_sched_t *sched;
2585 struct list_head enomem_conns;
2590 unsigned long deadline = jiffies;
2592 kportal_daemonize ("ksocknal_reaper");
2593 kportal_blockallsigs ();
2595 INIT_LIST_HEAD(&enomem_conns);
2596 init_waitqueue_entry (&wait, current);
2598 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2600 while (!ksocknal_data.ksnd_shuttingdown) {
2602 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2603 conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2604 ksock_conn_t, ksnc_list);
2605 list_del (&conn->ksnc_list);
2607 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2609 ksocknal_terminate_conn (conn);
2610 ksocknal_put_conn (conn);
2612 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2616 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2617 conn = list_entry (ksocknal_data.ksnd_zombie_conns.next,
2618 ksock_conn_t, ksnc_list);
2619 list_del (&conn->ksnc_list);
2621 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2623 ksocknal_destroy_conn (conn);
2625 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2629 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2630 list_add(&enomem_conns, &ksocknal_data.ksnd_enomem_conns);
2631 list_del_init(&ksocknal_data.ksnd_enomem_conns);
2634 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2636 /* reschedule all the connections that stalled with ENOMEM... */
2638 while (!list_empty (&enomem_conns)) {
2639 conn = list_entry (enomem_conns.next,
2640 ksock_conn_t, ksnc_tx_list);
2641 list_del (&conn->ksnc_tx_list);
2643 sched = conn->ksnc_scheduler;
2645 spin_lock_irqsave (&sched->kss_lock, flags);
2647 LASSERT (conn->ksnc_tx_scheduled);
2648 conn->ksnc_tx_ready = 1;
2649 list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
2650 wake_up (&sched->kss_waitq);
2652 spin_unlock_irqrestore (&sched->kss_lock, flags);
2656 /* careful with the jiffy wrap... */
2657 while ((timeout = (int)(deadline - jiffies)) <= 0) {
2660 int chunk = ksocknal_data.ksnd_peer_hash_size;
2662 /* Time to check for timeouts on a few more peers: I do
2663 * checks every 'p' seconds on a proportion of the peer
2664 * table and I need to check every connection 'n' times
2665 * within a timeout interval, to ensure I detect a
2666 * timeout on any connection within (n+1)/n times the
2667 * timeout interval. */
2669 if (ksocknal_data.ksnd_io_timeout > n * p)
2670 chunk = (chunk * n * p) /
2671 ksocknal_data.ksnd_io_timeout;
2675 for (i = 0; i < chunk; i++) {
2676 ksocknal_check_peer_timeouts (peer_index);
2677 peer_index = (peer_index + 1) %
2678 ksocknal_data.ksnd_peer_hash_size;
2684 if (nenomem_conns != 0) {
2685 /* Reduce my timeout if I rescheduled ENOMEM conns.
2686 * This also prevents me getting woken immediately
2687 * if any go back on my enomem list. */
2688 timeout = SOCKNAL_ENOMEM_RETRY;
2690 ksocknal_data.ksnd_reaper_waketime = jiffies + timeout;
2692 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2693 set_current_state (TASK_INTERRUPTIBLE);
2695 if (!ksocknal_data.ksnd_shuttingdown &&
2696 list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2697 list_empty (&ksocknal_data.ksnd_zombie_conns))
2698 schedule_timeout (timeout);
2700 set_current_state (TASK_RUNNING);
2701 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2703 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2706 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2708 ksocknal_thread_fini ();
2712 nal_cb_t ksocknal_lib = {
2713 nal_data: &ksocknal_data, /* NAL private data */
2714 cb_send: ksocknal_send,
2715 cb_send_pages: ksocknal_send_pages,
2716 cb_recv: ksocknal_recv,
2717 cb_recv_pages: ksocknal_recv_pages,
2718 cb_read: ksocknal_read,
2719 cb_write: ksocknal_write,
2720 cb_malloc: ksocknal_malloc,
2721 cb_free: ksocknal_free,
2722 cb_printf: ksocknal_printf,
2723 cb_cli: ksocknal_cli,
2724 cb_sti: ksocknal_sti,
2725 cb_dist: ksocknal_dist