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.
27 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
28 # include <linux/syscalls.h>
32 * LIB functions follow
36 ksocknal_dist(lib_nal_t *nal, ptl_nid_t nid, unsigned long *dist)
38 /* I would guess that if ksocknal_get_peer (nid) == NULL,
39 and we're not routing, then 'nid' is very distant :) */
40 if (nal->libnal_ni.ni_pid.nid == nid) {
50 ksocknal_free_ltx (ksock_ltx_t *ltx)
52 atomic_dec(&ksocknal_data.ksnd_nactive_ltxs);
53 PORTAL_FREE(ltx, ltx->ltx_desc_size);
56 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
58 ksocknal_kvaddr_to_page (unsigned long vaddr)
62 if (vaddr >= VMALLOC_START &&
64 page = vmalloc_to_page ((void *)vaddr);
66 else if (vaddr >= PKMAP_BASE &&
67 vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
68 page = vmalloc_to_page ((void *)vaddr);
69 /* in 2.4 ^ just walks the page tables */
72 page = virt_to_page (vaddr);
83 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
85 struct socket *sock = conn->ksnc_sock;
86 struct iovec *iov = tx->tx_iov;
87 int fragsize = iov->iov_len;
88 unsigned long vaddr = (unsigned long)iov->iov_base;
89 int more = (tx->tx_niov > 1) ||
91 (!list_empty (&conn->ksnc_tx_queue));
92 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
93 int offset = vaddr & (PAGE_SIZE - 1);
94 int zcsize = MIN (fragsize, PAGE_SIZE - offset);
99 /* NB we can't trust socket ops to either consume our iovs
100 * or leave them alone, so we only send 1 frag at a time. */
101 LASSERT (fragsize <= tx->tx_resid);
102 LASSERT (tx->tx_niov > 0);
104 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
105 if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
106 (sock->sk->route_caps & NETIF_F_SG) &&
107 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
108 (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
110 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
111 (void *)vaddr, page, page_address(page), offset, zcsize);
113 if (fragsize > zcsize) {
118 rc = tcp_sendpage_zccd(sock, page, offset, zcsize,
119 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
124 /* NB don't pass tx's iov; sendmsg may or may not update it */
125 struct iovec fragiov = { .iov_base = (void *)vaddr,
126 .iov_len = fragsize};
127 struct msghdr msg = {
134 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
136 mm_segment_t oldmm = get_fs();
139 rc = sock_sendmsg(sock, &msg, fragsize);
146 if (rc < iov->iov_len) {
147 /* didn't send whole iov entry... */
148 iov->iov_base = (void *)(vaddr + rc);
160 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
162 struct socket *sock = conn->ksnc_sock;
163 ptl_kiov_t *kiov = tx->tx_kiov;
164 int fragsize = kiov->kiov_len;
165 struct page *page = kiov->kiov_page;
166 int offset = kiov->kiov_offset;
167 int more = (tx->tx_nkiov > 1) ||
168 (!list_empty (&conn->ksnc_tx_queue));
171 /* NB we can't trust socket ops to either consume our iovs
172 * or leave them alone, so we only send 1 frag at a time. */
173 LASSERT (fragsize <= tx->tx_resid);
174 LASSERT (offset + fragsize <= PAGE_SIZE);
175 LASSERT (tx->tx_niov == 0);
176 LASSERT (tx->tx_nkiov > 0);
179 if (fragsize >= ksocknal_tunables.ksnd_zc_min_frag &&
180 (sock->sk->route_caps & NETIF_F_SG) &&
181 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
183 CDEBUG(D_NET, "page %p + offset %x for %d\n",
184 page, offset, fragsize);
186 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
187 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
192 char *addr = ((char *)kmap (page)) + offset;
193 struct iovec fragiov = {.iov_base = addr,
194 .iov_len = fragsize};
195 struct msghdr msg = {
202 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
204 mm_segment_t oldmm = get_fs();
207 rc = sock_sendmsg(sock, &msg, fragsize);
217 kiov->kiov_offset = offset + rc;
218 kiov->kiov_len = fragsize - rc;
229 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
234 if (ksocknal_data.ksnd_stall_tx != 0) {
235 set_current_state (TASK_UNINTERRUPTIBLE);
236 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
239 LASSERT (tx->tx_resid != 0);
241 rc = ksocknal_getconnsock (conn);
243 LASSERT (conn->ksnc_closing);
248 if (ksocknal_data.ksnd_enomem_tx > 0) {
250 ksocknal_data.ksnd_enomem_tx--;
252 } else if (tx->tx_niov != 0) {
253 rc = ksocknal_send_iov (conn, tx);
255 rc = ksocknal_send_kiov (conn, tx);
258 bufnob = conn->ksnc_sock->sk->sk_wmem_queued;
259 if (rc > 0) /* sent something? */
260 conn->ksnc_tx_bufnob += rc; /* account it */
262 if (bufnob < conn->ksnc_tx_bufnob) {
263 /* allocated send buffer bytes < computed; infer
264 * something got ACKed */
265 conn->ksnc_tx_deadline = jiffies +
266 ksocknal_tunables.ksnd_io_timeout * HZ;
267 conn->ksnc_peer->ksnp_last_alive = jiffies;
268 conn->ksnc_tx_bufnob = bufnob;
273 /* Didn't write anything.
275 * NB: rc == 0 and rc == -EAGAIN both mean try
276 * again later (linux stack returns -EAGAIN for
277 * this, but Adaptech TOE returns 0).
279 * Also, sends never fail with -ENOMEM, just
280 * -EAGAIN, but with the added bonus that we can't
281 * expect write_space() to call us back to tell us
282 * when to try sending again. We use the
283 * SOCK_NOSPACE flag to diagnose... */
285 LASSERT(rc != -ENOMEM);
287 if (rc == 0 || rc == -EAGAIN) {
288 if (test_bit(SOCK_NOSPACE,
289 &conn->ksnc_sock->flags)) {
295 if ((counter & (-counter)) == counter)
296 CWARN("%d ENOMEM tx %p\n",
304 /* socket's wmem_queued now includes 'rc' bytes */
305 atomic_sub (rc, &conn->ksnc_tx_nob);
308 } while (tx->tx_resid != 0);
310 ksocknal_putconnsock (conn);
315 ksocknal_eager_ack (ksock_conn_t *conn)
318 mm_segment_t oldmm = get_fs();
319 struct socket *sock = conn->ksnc_sock;
321 /* Remind the socket to ACK eagerly. If I don't, the socket might
322 * think I'm about to send something it could piggy-back the ACK
323 * on, introducing delay in completing zero-copy sends in my
327 sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
328 (char *)&opt, sizeof (opt));
333 ksocknal_recv_iov (ksock_conn_t *conn)
335 struct iovec *iov = conn->ksnc_rx_iov;
336 int fragsize = iov->iov_len;
337 unsigned long vaddr = (unsigned long)iov->iov_base;
338 struct iovec fragiov = { .iov_base = (void *)vaddr,
339 .iov_len = fragsize};
340 struct msghdr msg = {
349 mm_segment_t oldmm = get_fs();
352 /* NB we can't trust socket ops to either consume our iovs
353 * or leave them alone, so we only receive 1 frag at a time. */
354 LASSERT (conn->ksnc_rx_niov > 0);
355 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
358 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
359 /* NB this is just a boolean............................^ */
365 /* received something... */
366 conn->ksnc_peer->ksnp_last_alive = jiffies;
367 conn->ksnc_rx_deadline = jiffies +
368 ksocknal_tunables.ksnd_io_timeout * HZ;
369 mb(); /* order with setting rx_started */
370 conn->ksnc_rx_started = 1;
372 conn->ksnc_rx_nob_wanted -= rc;
373 conn->ksnc_rx_nob_left -= rc;
376 iov->iov_base = (void *)(vaddr + rc);
377 iov->iov_len = fragsize - rc;
382 conn->ksnc_rx_niov--;
387 ksocknal_recv_kiov (ksock_conn_t *conn)
389 ptl_kiov_t *kiov = conn->ksnc_rx_kiov;
390 struct page *page = kiov->kiov_page;
391 int offset = kiov->kiov_offset;
392 int fragsize = kiov->kiov_len;
393 unsigned long vaddr = ((unsigned long)kmap (page)) + offset;
394 struct iovec fragiov = { .iov_base = (void *)vaddr,
395 .iov_len = fragsize};
396 struct msghdr msg = {
405 mm_segment_t oldmm = get_fs();
408 /* NB we can't trust socket ops to either consume our iovs
409 * or leave them alone, so we only receive 1 frag at a time. */
410 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
411 LASSERT (conn->ksnc_rx_nkiov > 0);
412 LASSERT (offset + fragsize <= PAGE_SIZE);
415 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
416 /* NB this is just a boolean............................^ */
424 /* received something... */
425 conn->ksnc_peer->ksnp_last_alive = jiffies;
426 conn->ksnc_rx_deadline = jiffies +
427 ksocknal_tunables.ksnd_io_timeout * HZ;
428 mb(); /* order with setting rx_started */
429 conn->ksnc_rx_started = 1;
431 conn->ksnc_rx_nob_wanted -= rc;
432 conn->ksnc_rx_nob_left -= rc;
435 kiov->kiov_offset = offset + rc;
436 kiov->kiov_len = fragsize - rc;
440 conn->ksnc_rx_kiov++;
441 conn->ksnc_rx_nkiov--;
446 ksocknal_receive (ksock_conn_t *conn)
448 /* Return 1 on success, 0 on EOF, < 0 on error.
449 * Caller checks ksnc_rx_nob_wanted to determine
450 * progress/completion. */
454 if (ksocknal_data.ksnd_stall_rx != 0) {
455 set_current_state (TASK_UNINTERRUPTIBLE);
456 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
459 rc = ksocknal_getconnsock (conn);
461 LASSERT (conn->ksnc_closing);
466 if (conn->ksnc_rx_niov != 0)
467 rc = ksocknal_recv_iov (conn);
469 rc = ksocknal_recv_kiov (conn);
472 /* error/EOF or partial receive */
475 } else if (rc == 0 && conn->ksnc_rx_started) {
476 /* EOF in the middle of a message */
482 /* Completed a fragment */
484 if (conn->ksnc_rx_nob_wanted == 0) {
485 /* Completed a message segment (header or payload) */
486 if ((ksocknal_tunables.ksnd_eager_ack & conn->ksnc_type) != 0 &&
487 (conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
488 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
489 /* Remind the socket to ack eagerly... */
490 ksocknal_eager_ack(conn);
497 ksocknal_putconnsock (conn);
503 ksocknal_zc_callback (zccd_t *zcd)
505 ksock_tx_t *tx = KSOCK_ZCCD_2_TX(zcd);
506 ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
510 /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
512 spin_lock_irqsave (&sched->kss_lock, flags);
514 list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
515 wake_up (&sched->kss_waitq);
517 spin_unlock_irqrestore (&sched->kss_lock, flags);
523 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
528 if (tx->tx_conn != NULL) {
530 /* zero copy completion isn't always from
531 * process_transmit() so it needs to keep a ref on
534 ksocknal_put_conn (tx->tx_conn);
540 if (tx->tx_isfwd) { /* was a forwarded packet? */
541 kpr_fwd_done (&ksocknal_data.ksnd_router,
542 KSOCK_TX_2_KPR_FWD_DESC (tx),
543 (tx->tx_resid == 0) ? 0 : -ECONNABORTED);
549 ltx = KSOCK_TX_2_KSOCK_LTX (tx);
551 lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie,
552 (tx->tx_resid == 0) ? PTL_OK : PTL_FAIL);
554 ksocknal_free_ltx (ltx);
559 ksocknal_tx_launched (ksock_tx_t *tx)
562 if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
563 ksock_conn_t *conn = tx->tx_conn;
565 /* zccd skbufs are still in-flight. First take a ref on
566 * conn, so it hangs about for ksocknal_tx_done... */
567 atomic_inc (&conn->ksnc_refcount);
569 /* ...then drop the initial ref on zccd, so the zero copy
570 * callback can occur */
571 zccd_put (&tx->tx_zccd);
575 /* Any zero-copy-ness (if any) has completed; I can complete the
576 * transmit now, avoiding an extra schedule */
577 ksocknal_tx_done (tx, 0);
581 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
586 rc = ksocknal_transmit (conn, tx);
588 CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
590 if (tx->tx_resid == 0) {
591 /* Sent everything OK */
594 ksocknal_tx_launched (tx);
602 /* Queue on ksnd_enomem_conns for retry after a timeout */
603 spin_lock_irqsave(&ksocknal_data.ksnd_reaper_lock, flags);
605 /* enomem list takes over scheduler's ref... */
606 LASSERT (conn->ksnc_tx_scheduled);
607 list_add_tail(&conn->ksnc_tx_list,
608 &ksocknal_data.ksnd_enomem_conns);
609 if (!time_after_eq(jiffies + SOCKNAL_ENOMEM_RETRY,
610 ksocknal_data.ksnd_reaper_waketime))
611 wake_up (&ksocknal_data.ksnd_reaper_waitq);
613 spin_unlock_irqrestore(&ksocknal_data.ksnd_reaper_lock, flags);
620 if (!conn->ksnc_closing)
621 CERROR("[%p] Error %d on write to "LPX64
622 " ip %d.%d.%d.%d:%d\n", conn, rc,
623 conn->ksnc_peer->ksnp_nid,
624 HIPQUAD(conn->ksnc_ipaddr),
627 ksocknal_close_conn_and_siblings (conn, rc);
628 ksocknal_tx_launched (tx);
634 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
638 /* called holding write lock on ksnd_global_lock */
640 LASSERT (!route->ksnr_deleted);
641 LASSERT ((route->ksnr_connected & (1 << SOCKNAL_CONN_ANY)) == 0);
642 LASSERT ((route->ksnr_connected & KSNR_TYPED_ROUTES) != KSNR_TYPED_ROUTES);
643 LASSERT (route->ksnr_connecting == 0);
645 if (ksocknal_tunables.ksnd_typed_conns)
646 route->ksnr_connecting =
647 KSNR_TYPED_ROUTES & ~route->ksnr_connected;
649 route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
651 atomic_inc (&route->ksnr_refcount); /* extra ref for asynchd */
653 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
655 list_add_tail (&route->ksnr_connect_list,
656 &ksocknal_data.ksnd_autoconnectd_routes);
657 wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
659 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
663 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
665 char ipbuf[PTL_NALFMT_SIZE];
666 ptl_nid_t target_nid;
668 ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
674 CERROR ("Can't send packet to "LPX64
675 " %s: routed target is not a peer\n",
676 nid, portals_nid2str(SOCKNAL, nid, ipbuf));
680 rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
683 CERROR ("Can't route to "LPX64" %s: router error %d\n",
684 nid, portals_nid2str(SOCKNAL, nid, ipbuf), rc);
688 peer = ksocknal_find_peer_locked (target_nid);
692 CERROR ("Can't send packet to "LPX64" %s: no peer entry\n",
693 target_nid, portals_nid2str(SOCKNAL, target_nid, ipbuf));
698 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
700 struct list_head *tmp;
701 ksock_conn_t *typed = NULL;
703 ksock_conn_t *fallback = NULL;
707 list_for_each (tmp, &peer->ksnp_conns) {
708 ksock_conn_t *c = list_entry(tmp, ksock_conn_t, ksnc_list);
709 #if SOCKNAL_ROUND_ROBIN
712 int nob = atomic_read(&c->ksnc_tx_nob) +
713 c->ksnc_sock->sk->sk_wmem_queued;
715 LASSERT (!c->ksnc_closing);
717 if (fallback == NULL || nob < fnob) {
722 if (!ksocknal_tunables.ksnd_typed_conns)
725 switch (c->ksnc_type) {
728 case SOCKNAL_CONN_ANY:
730 case SOCKNAL_CONN_BULK_IN:
732 case SOCKNAL_CONN_BULK_OUT:
733 if (tx->tx_nob < ksocknal_tunables.ksnd_min_bulk)
736 case SOCKNAL_CONN_CONTROL:
737 if (tx->tx_nob >= ksocknal_tunables.ksnd_min_bulk)
742 if (typed == NULL || nob < tnob) {
748 /* prefer the typed selection */
749 conn = (typed != NULL) ? typed : fallback;
751 #if SOCKNAL_ROUND_ROBIN
753 /* round-robin all else being equal */
754 list_del (&conn->ksnc_list);
755 list_add_tail (&conn->ksnc_list, &peer->ksnp_conns);
762 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
765 ksock_sched_t *sched = conn->ksnc_scheduler;
767 /* called holding global lock (read or irq-write) and caller may
768 * not have dropped this lock between finding conn and calling me,
769 * so we don't need the {get,put}connsock dance to deref
771 LASSERT(!conn->ksnc_closing);
772 LASSERT(tx->tx_resid == tx->tx_nob);
774 CDEBUG (D_NET, "Sending to "LPX64" ip %d.%d.%d.%d:%d\n",
775 conn->ksnc_peer->ksnp_nid,
776 HIPQUAD(conn->ksnc_ipaddr),
779 atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
783 zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
784 /* NB this sets 1 ref on zccd, so the callback can only occur after
785 * I've released this ref. */
787 spin_lock_irqsave (&sched->kss_lock, flags);
789 if (list_empty(&conn->ksnc_tx_queue) &&
790 conn->ksnc_sock->sk->sk_wmem_queued == 0) {
791 /* First packet starts the timeout */
792 conn->ksnc_tx_deadline = jiffies +
793 ksocknal_tunables.ksnd_io_timeout * HZ;
794 conn->ksnc_tx_bufnob = 0;
795 mb(); /* order with adding to tx_queue */
798 list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
800 if (conn->ksnc_tx_ready && /* able to send */
801 !conn->ksnc_tx_scheduled) { /* not scheduled to send */
802 /* +1 ref for scheduler */
803 atomic_inc (&conn->ksnc_refcount);
804 list_add_tail (&conn->ksnc_tx_list,
805 &sched->kss_tx_conns);
806 conn->ksnc_tx_scheduled = 1;
807 wake_up (&sched->kss_waitq);
810 spin_unlock_irqrestore (&sched->kss_lock, flags);
814 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
816 struct list_head *tmp;
817 ksock_route_t *route;
820 list_for_each (tmp, &peer->ksnp_routes) {
821 route = list_entry (tmp, ksock_route_t, ksnr_list);
822 bits = route->ksnr_connected;
824 /* All typed connections established? */
825 if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES)
828 /* Untyped connection established? */
829 if ((bits & (1 << SOCKNAL_CONN_ANY)) != 0)
832 /* connection being established? */
833 if (route->ksnr_connecting != 0)
836 /* too soon to retry this guy? */
837 if (!time_after_eq (jiffies, route->ksnr_timeout))
847 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
849 struct list_head *tmp;
850 ksock_route_t *route;
852 list_for_each (tmp, &peer->ksnp_routes) {
853 route = list_entry (tmp, ksock_route_t, ksnr_list);
855 if (route->ksnr_connecting != 0)
863 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
868 ksock_route_t *route;
871 /* Ensure the frags we've been given EXACTLY match the number of
872 * bytes we want to send. Many TCP/IP stacks disregard any total
873 * size parameters passed to them and just look at the frags.
875 * We always expect at least 1 mapped fragment containing the
876 * complete portals header. */
877 LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
878 lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
879 LASSERT (tx->tx_niov >= 1);
880 LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
882 CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
883 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type,
884 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
886 tx->tx_conn = NULL; /* only set when assigned a conn */
887 tx->tx_resid = tx->tx_nob;
888 tx->tx_hdr = (ptl_hdr_t *)tx->tx_iov[0].iov_base;
890 g_lock = &ksocknal_data.ksnd_global_lock;
891 #if !SOCKNAL_ROUND_ROBIN
894 peer = ksocknal_find_target_peer_locked (tx, nid);
896 read_unlock (g_lock);
897 return (-EHOSTUNREACH);
900 if (ksocknal_find_connectable_route_locked(peer) == NULL) {
901 conn = ksocknal_find_conn_locked (tx, peer);
903 /* I've got no autoconnect routes that need to be
904 * connecting and I do have an actual connection... */
905 ksocknal_queue_tx_locked (tx, conn);
906 read_unlock (g_lock);
911 /* I'll need a write lock... */
912 read_unlock (g_lock);
914 write_lock_irqsave(g_lock, flags);
916 peer = ksocknal_find_target_peer_locked (tx, nid);
918 write_unlock_irqrestore(g_lock, flags);
919 return (-EHOSTUNREACH);
923 /* launch any/all autoconnections that need it */
924 route = ksocknal_find_connectable_route_locked (peer);
928 ksocknal_launch_autoconnect_locked (route);
931 conn = ksocknal_find_conn_locked (tx, peer);
933 /* Connection exists; queue message on it */
934 ksocknal_queue_tx_locked (tx, conn);
935 write_unlock_irqrestore (g_lock, flags);
939 route = ksocknal_find_connecting_route_locked (peer);
941 /* At least 1 connection is being established; queue the
943 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
944 write_unlock_irqrestore (g_lock, flags);
948 write_unlock_irqrestore (g_lock, flags);
949 return (-EHOSTUNREACH);
953 ksocknal_sendmsg(lib_nal_t *nal,
960 unsigned int payload_niov,
961 struct iovec *payload_iov,
962 ptl_kiov_t *payload_kiov,
963 size_t payload_offset,
970 /* NB 'private' is different depending on what we're sending.
971 * Just ignore it... */
973 CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid:"LPX64
974 " pid %d\n", payload_nob, payload_niov, nid , pid);
976 LASSERT (payload_nob == 0 || payload_niov > 0);
977 LASSERT (payload_niov <= PTL_MD_MAX_IOV);
979 /* It must be OK to kmap() if required */
980 LASSERT (payload_kiov == NULL || !in_interrupt ());
981 /* payload is either all vaddrs or all pages */
982 LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
984 if (payload_iov != NULL)
985 desc_size = offsetof(ksock_ltx_t, ltx_iov[1 + payload_niov]);
987 desc_size = offsetof(ksock_ltx_t, ltx_kiov[payload_niov]);
989 if (in_interrupt() ||
990 type == PTL_MSG_ACK ||
991 type == PTL_MSG_REPLY) {
992 /* Can't block if in interrupt or responding to an incoming
994 PORTAL_ALLOC_ATOMIC(ltx, desc_size);
996 PORTAL_ALLOC(ltx, desc_size);
1000 CERROR("Can't allocate tx desc type %d size %d %s\n",
1001 type, desc_size, in_interrupt() ? "(intr)" : "");
1002 return (PTL_NO_SPACE);
1005 atomic_inc(&ksocknal_data.ksnd_nactive_ltxs);
1007 ltx->ltx_desc_size = desc_size;
1009 /* We always have 1 mapped frag for the header */
1010 ltx->ltx_tx.tx_iov = ltx->ltx_iov;
1011 ltx->ltx_iov[0].iov_base = <x->ltx_hdr;
1012 ltx->ltx_iov[0].iov_len = sizeof(*hdr);
1013 ltx->ltx_hdr = *hdr;
1015 ltx->ltx_private = private;
1016 ltx->ltx_cookie = cookie;
1018 ltx->ltx_tx.tx_isfwd = 0;
1019 ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_nob;
1021 if (payload_iov != NULL) {
1022 /* payload is all mapped */
1023 ltx->ltx_tx.tx_kiov = NULL;
1024 ltx->ltx_tx.tx_nkiov = 0;
1026 ltx->ltx_tx.tx_niov =
1027 1 + lib_extract_iov(payload_niov, <x->ltx_iov[1],
1028 payload_niov, payload_iov,
1029 payload_offset, payload_nob);
1031 /* payload is all pages */
1032 ltx->ltx_tx.tx_niov = 1;
1034 ltx->ltx_tx.tx_kiov = ltx->ltx_kiov;
1035 ltx->ltx_tx.tx_nkiov =
1036 lib_extract_kiov(payload_niov, ltx->ltx_kiov,
1037 payload_niov, payload_kiov,
1038 payload_offset, payload_nob);
1041 rc = ksocknal_launch_packet(<x->ltx_tx, nid);
1045 ksocknal_free_ltx(ltx);
1050 ksocknal_send (lib_nal_t *nal, void *private, lib_msg_t *cookie,
1051 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1052 unsigned int payload_niov, struct iovec *payload_iov,
1053 size_t payload_offset, size_t payload_len)
1055 return (ksocknal_sendmsg(nal, private, cookie,
1056 hdr, type, nid, pid,
1057 payload_niov, payload_iov, NULL,
1058 payload_offset, payload_len));
1062 ksocknal_send_pages (lib_nal_t *nal, void *private, lib_msg_t *cookie,
1063 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1064 unsigned int payload_niov, ptl_kiov_t *payload_kiov,
1065 size_t payload_offset, size_t payload_len)
1067 return (ksocknal_sendmsg(nal, private, cookie,
1068 hdr, type, nid, pid,
1069 payload_niov, NULL, payload_kiov,
1070 payload_offset, payload_len));
1074 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1076 ptl_nid_t nid = fwd->kprfd_gateway_nid;
1077 ksock_ftx_t *ftx = (ksock_ftx_t *)&fwd->kprfd_scratch;
1080 CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1081 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1083 /* I'm the gateway; must be the last hop */
1084 if (nid == ksocknal_lib.libnal_ni.ni_pid.nid)
1085 nid = fwd->kprfd_target_nid;
1087 /* setup iov for hdr */
1088 ftx->ftx_iov.iov_base = fwd->kprfd_hdr;
1089 ftx->ftx_iov.iov_len = sizeof(ptl_hdr_t);
1091 ftx->ftx_tx.tx_isfwd = 1; /* This is a forwarding packet */
1092 ftx->ftx_tx.tx_nob = sizeof(ptl_hdr_t) + fwd->kprfd_nob;
1093 ftx->ftx_tx.tx_niov = 1;
1094 ftx->ftx_tx.tx_iov = &ftx->ftx_iov;
1095 ftx->ftx_tx.tx_nkiov = fwd->kprfd_niov;
1096 ftx->ftx_tx.tx_kiov = fwd->kprfd_kiov;
1098 rc = ksocknal_launch_packet (&ftx->ftx_tx, nid);
1100 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
1104 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1106 long pid = kernel_thread (fn, arg, 0);
1107 unsigned long flags;
1112 write_lock_irqsave(&ksocknal_data.ksnd_global_lock, flags);
1113 ksocknal_data.ksnd_nthreads++;
1114 write_unlock_irqrestore(&ksocknal_data.ksnd_global_lock, flags);
1119 ksocknal_thread_fini (void)
1121 unsigned long flags;
1123 write_lock_irqsave(&ksocknal_data.ksnd_global_lock, flags);
1124 ksocknal_data.ksnd_nthreads--;
1125 write_unlock_irqrestore(&ksocknal_data.ksnd_global_lock, flags);
1129 ksocknal_fmb_callback (void *arg, int error)
1131 ksock_fmb_t *fmb = (ksock_fmb_t *)arg;
1132 ksock_fmb_pool_t *fmp = fmb->fmb_pool;
1133 ptl_hdr_t *hdr = &fmb->fmb_hdr;
1134 ksock_conn_t *conn = NULL;
1135 ksock_sched_t *sched;
1136 unsigned long flags;
1137 char ipbuf[PTL_NALFMT_SIZE];
1138 char ipbuf2[PTL_NALFMT_SIZE];
1141 CERROR("Failed to route packet from "
1142 LPX64" %s to "LPX64" %s: %d\n",
1143 le64_to_cpu(hdr->src_nid),
1144 portals_nid2str(SOCKNAL, le64_to_cpu(hdr->src_nid), ipbuf),
1145 le64_to_cpu(hdr->dest_nid),
1146 portals_nid2str(SOCKNAL, le64_to_cpu(hdr->dest_nid), ipbuf2),
1149 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1150 le64_to_cpu(hdr->src_nid), le64_to_cpu(hdr->dest_nid));
1152 /* drop peer ref taken on init */
1153 ksocknal_put_peer (fmb->fmb_peer);
1155 spin_lock_irqsave (&fmp->fmp_lock, flags);
1157 list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1158 fmp->fmp_nactive_fmbs--;
1160 if (!list_empty (&fmp->fmp_blocked_conns)) {
1161 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1162 ksock_conn_t, ksnc_rx_list);
1163 list_del (&conn->ksnc_rx_list);
1166 spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1171 CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1172 LASSERT (conn->ksnc_rx_scheduled);
1173 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1175 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1177 sched = conn->ksnc_scheduler;
1179 spin_lock_irqsave (&sched->kss_lock, flags);
1181 list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1182 wake_up (&sched->kss_waitq);
1184 spin_unlock_irqrestore (&sched->kss_lock, flags);
1188 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1190 int payload_nob = conn->ksnc_rx_nob_left;
1191 unsigned long flags;
1192 ksock_fmb_pool_t *pool;
1195 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1196 LASSERT (kpr_routing(&ksocknal_data.ksnd_router));
1198 if (payload_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1199 pool = &ksocknal_data.ksnd_small_fmp;
1201 pool = &ksocknal_data.ksnd_large_fmp;
1203 spin_lock_irqsave (&pool->fmp_lock, flags);
1205 if (!list_empty (&pool->fmp_idle_fmbs)) {
1206 fmb = list_entry(pool->fmp_idle_fmbs.next,
1207 ksock_fmb_t, fmb_list);
1208 list_del (&fmb->fmb_list);
1209 pool->fmp_nactive_fmbs++;
1210 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1215 /* deschedule until fmb free */
1217 conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1219 list_add_tail (&conn->ksnc_rx_list,
1220 &pool->fmp_blocked_conns);
1222 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1227 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1229 int payload_nob = conn->ksnc_rx_nob_left;
1230 ptl_nid_t dest_nid = le64_to_cpu(conn->ksnc_hdr.dest_nid);
1232 int nob = payload_nob;
1234 LASSERT (conn->ksnc_rx_scheduled);
1235 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1236 LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1237 LASSERT (payload_nob >= 0);
1238 LASSERT (payload_nob <= fmb->fmb_pool->fmp_buff_pages * PAGE_SIZE);
1239 LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1240 LASSERT (fmb->fmb_kiov[0].kiov_offset == 0);
1242 /* Take a ref on the conn's peer to prevent module unload before
1243 * forwarding completes. */
1244 fmb->fmb_peer = conn->ksnc_peer;
1245 atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1247 /* Copy the header we just read into the forwarding buffer. If
1248 * there's payload, start reading reading it into the buffer,
1249 * otherwise the forwarding buffer can be kicked off
1251 fmb->fmb_hdr = conn->ksnc_hdr;
1254 LASSERT (niov < fmb->fmb_pool->fmp_buff_pages);
1255 LASSERT (fmb->fmb_kiov[niov].kiov_offset == 0);
1256 fmb->fmb_kiov[niov].kiov_len = MIN (PAGE_SIZE, nob);
1261 kpr_fwd_init(&fmb->fmb_fwd, dest_nid, &fmb->fmb_hdr,
1262 payload_nob, niov, fmb->fmb_kiov,
1263 ksocknal_fmb_callback, fmb);
1265 if (payload_nob == 0) { /* got complete packet already */
1266 CDEBUG (D_NET, "%p "LPX64"->"LPX64" fwd_start (immediate)\n",
1267 conn, le64_to_cpu(conn->ksnc_hdr.src_nid), dest_nid);
1269 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1271 ksocknal_new_packet (conn, 0); /* on to next packet */
1275 conn->ksnc_cookie = fmb; /* stash fmb for later */
1276 conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1278 /* Set up conn->ksnc_rx_kiov to read the payload into fmb's kiov-ed
1280 LASSERT (niov <= sizeof(conn->ksnc_rx_iov_space)/sizeof(ptl_kiov_t));
1282 conn->ksnc_rx_niov = 0;
1283 conn->ksnc_rx_nkiov = niov;
1284 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1285 memcpy(conn->ksnc_rx_kiov, fmb->fmb_kiov, niov * sizeof(ptl_kiov_t));
1287 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1288 le64_to_cpu(conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1293 ksocknal_fwd_parse (ksock_conn_t *conn)
1296 ptl_nid_t dest_nid = le64_to_cpu(conn->ksnc_hdr.dest_nid);
1297 ptl_nid_t src_nid = le64_to_cpu(conn->ksnc_hdr.src_nid);
1298 int body_len = le32_to_cpu(conn->ksnc_hdr.payload_length);
1299 char str[PTL_NALFMT_SIZE];
1300 char str2[PTL_NALFMT_SIZE];
1302 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1303 src_nid, dest_nid, conn->ksnc_rx_nob_left);
1305 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1306 LASSERT (conn->ksnc_rx_scheduled);
1308 if (body_len < 0) { /* length corrupt (overflow) */
1309 CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
1310 "packet size %d illegal\n",
1311 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1312 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1315 ksocknal_new_packet (conn, 0); /* on to new packet */
1319 if (!kpr_routing(&ksocknal_data.ksnd_router)) { /* not forwarding */
1320 CERROR("dropping packet from "LPX64" (%s) for "LPX64
1321 " (%s): not forwarding\n",
1322 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1323 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1324 /* on to new packet (skip this one's body) */
1325 ksocknal_new_packet (conn, body_len);
1329 if (body_len > PTL_MTU) { /* too big to forward */
1330 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1331 "(%s): packet size %d too big\n",
1332 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1333 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1335 /* on to new packet (skip this one's body) */
1336 ksocknal_new_packet (conn, body_len);
1340 /* should have gone direct */
1341 peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1343 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1344 "(%s): target is a peer\n",
1345 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1346 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1347 ksocknal_put_peer (peer); /* drop ref from get above */
1349 /* on to next packet (skip this one's body) */
1350 ksocknal_new_packet (conn, body_len);
1354 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB; /* Getting FMB now */
1355 conn->ksnc_rx_nob_left = body_len; /* stash packet size */
1356 conn->ksnc_rx_nob_wanted = body_len; /* (no slop) */
1360 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1362 static char ksocknal_slop_buffer[4096];
1368 if (nob_to_skip == 0) { /* right at next packet boundary now */
1369 conn->ksnc_rx_started = 0;
1370 mb (); /* racing with timeout thread */
1372 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1373 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1374 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1376 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1377 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1378 conn->ksnc_rx_iov[0].iov_len = sizeof (ptl_hdr_t);
1379 conn->ksnc_rx_niov = 1;
1381 conn->ksnc_rx_kiov = NULL;
1382 conn->ksnc_rx_nkiov = 0;
1386 /* Set up to skip as much a possible now. If there's more left
1387 * (ran out of iov entries) we'll get called again */
1389 conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1390 conn->ksnc_rx_nob_left = nob_to_skip;
1391 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1396 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1398 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1399 conn->ksnc_rx_iov[niov].iov_len = nob;
1404 } while (nob_to_skip != 0 && /* mustn't overflow conn's rx iov */
1405 niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1407 conn->ksnc_rx_niov = niov;
1408 conn->ksnc_rx_kiov = NULL;
1409 conn->ksnc_rx_nkiov = 0;
1410 conn->ksnc_rx_nob_wanted = skipped;
1415 ksocknal_process_receive (ksock_conn_t *conn)
1420 LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1422 /* doesn't need a forwarding buffer */
1423 if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1427 fmb = ksocknal_get_idle_fmb (conn);
1429 /* conn descheduled waiting for idle fmb */
1433 if (ksocknal_init_fmb (conn, fmb)) {
1434 /* packet forwarded */
1439 /* NB: sched lock NOT held */
1440 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1441 conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1442 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1443 conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1445 LASSERT (conn->ksnc_rx_nob_wanted > 0);
1447 rc = ksocknal_receive(conn);
1450 LASSERT (rc != -EAGAIN);
1453 CWARN ("[%p] EOF from "LPX64" ip %d.%d.%d.%d:%d\n",
1454 conn, conn->ksnc_peer->ksnp_nid,
1455 HIPQUAD(conn->ksnc_ipaddr),
1457 else if (!conn->ksnc_closing)
1458 CERROR ("[%p] Error %d on read from "LPX64
1459 " ip %d.%d.%d.%d:%d\n",
1460 conn, rc, conn->ksnc_peer->ksnp_nid,
1461 HIPQUAD(conn->ksnc_ipaddr),
1464 ksocknal_close_conn_and_siblings (conn, rc);
1465 return (rc == 0 ? -ESHUTDOWN : rc);
1468 if (conn->ksnc_rx_nob_wanted != 0) {
1473 switch (conn->ksnc_rx_state) {
1474 case SOCKNAL_RX_HEADER:
1475 if (conn->ksnc_hdr.type != cpu_to_le32(PTL_MSG_HELLO) &&
1476 le64_to_cpu(conn->ksnc_hdr.dest_nid) !=
1477 ksocknal_lib.libnal_ni.ni_pid.nid) {
1478 /* This packet isn't for me */
1479 ksocknal_fwd_parse (conn);
1480 switch (conn->ksnc_rx_state) {
1481 case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1482 return (0); /* => come back later */
1483 case SOCKNAL_RX_SLOP: /* skipping packet's body */
1484 goto try_read; /* => go read it */
1485 case SOCKNAL_RX_GET_FMB: /* forwarding */
1486 goto get_fmb; /* => go get a fwd msg buffer */
1493 /* sets wanted_len, iovs etc */
1494 rc = lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1497 /* I just received garbage: give up on this conn */
1498 ksocknal_close_conn_and_siblings (conn, rc);
1502 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1503 conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1504 goto try_read; /* go read the payload */
1506 /* Fall through (completed packet for me) */
1508 case SOCKNAL_RX_BODY:
1509 /* payload all received */
1510 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_OK);
1513 case SOCKNAL_RX_SLOP:
1514 /* starting new packet? */
1515 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1516 return (0); /* come back later */
1517 goto try_read; /* try to finish reading slop now */
1519 case SOCKNAL_RX_BODY_FWD:
1520 /* payload all received */
1521 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1522 conn, le64_to_cpu(conn->ksnc_hdr.src_nid),
1523 le64_to_cpu(conn->ksnc_hdr.dest_nid),
1524 conn->ksnc_rx_nob_left);
1526 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1527 * conn->ksnc_cookie */
1528 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1529 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1531 /* no slop in forwarded packets */
1532 LASSERT (conn->ksnc_rx_nob_left == 0);
1534 ksocknal_new_packet (conn, 0); /* on to next packet */
1535 return (0); /* (later) */
1543 return (-EINVAL); /* keep gcc happy */
1547 ksocknal_recv (lib_nal_t *nal, void *private, lib_msg_t *msg,
1548 unsigned int niov, struct iovec *iov,
1549 size_t offset, size_t mlen, size_t rlen)
1551 ksock_conn_t *conn = (ksock_conn_t *)private;
1553 LASSERT (mlen <= rlen);
1554 LASSERT (niov <= PTL_MD_MAX_IOV);
1556 conn->ksnc_cookie = msg;
1557 conn->ksnc_rx_nob_wanted = mlen;
1558 conn->ksnc_rx_nob_left = rlen;
1560 conn->ksnc_rx_nkiov = 0;
1561 conn->ksnc_rx_kiov = NULL;
1562 conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1563 conn->ksnc_rx_niov =
1564 lib_extract_iov(PTL_MD_MAX_IOV, conn->ksnc_rx_iov,
1565 niov, iov, offset, mlen);
1568 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1569 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1575 ksocknal_recv_pages (lib_nal_t *nal, void *private, lib_msg_t *msg,
1576 unsigned int niov, ptl_kiov_t *kiov,
1577 size_t offset, size_t mlen, size_t rlen)
1579 ksock_conn_t *conn = (ksock_conn_t *)private;
1581 LASSERT (mlen <= rlen);
1582 LASSERT (niov <= PTL_MD_MAX_IOV);
1584 conn->ksnc_cookie = msg;
1585 conn->ksnc_rx_nob_wanted = mlen;
1586 conn->ksnc_rx_nob_left = rlen;
1588 conn->ksnc_rx_niov = 0;
1589 conn->ksnc_rx_iov = NULL;
1590 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1591 conn->ksnc_rx_nkiov =
1592 lib_extract_kiov(PTL_MD_MAX_IOV, conn->ksnc_rx_kiov,
1593 niov, kiov, offset, mlen);
1596 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1597 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1603 ksocknal_sched_cansleep(ksock_sched_t *sched)
1605 unsigned long flags;
1608 spin_lock_irqsave(&sched->kss_lock, flags);
1610 rc = (!ksocknal_data.ksnd_shuttingdown &&
1612 list_empty(&sched->kss_zctxdone_list) &&
1614 list_empty(&sched->kss_rx_conns) &&
1615 list_empty(&sched->kss_tx_conns));
1617 spin_unlock_irqrestore(&sched->kss_lock, flags);
1621 int ksocknal_scheduler (void *arg)
1623 ksock_sched_t *sched = (ksock_sched_t *)arg;
1626 unsigned long flags;
1629 int id = sched - ksocknal_data.ksnd_schedulers;
1632 snprintf (name, sizeof (name),"ksocknald_%02d", id);
1633 kportal_daemonize (name);
1634 kportal_blockallsigs ();
1636 #if (CONFIG_SMP && CPU_AFFINITY)
1637 id = ksocknal_sched2cpu(id);
1638 if (cpu_online(id)) {
1641 set_cpus_allowed(current, m);
1643 CERROR ("Can't set CPU affinity for %s to %d\n", name, id);
1645 #endif /* CONFIG_SMP && CPU_AFFINITY */
1647 spin_lock_irqsave (&sched->kss_lock, flags);
1649 while (!ksocknal_data.ksnd_shuttingdown) {
1650 int did_something = 0;
1652 /* Ensure I progress everything semi-fairly */
1654 if (!list_empty (&sched->kss_rx_conns)) {
1655 conn = list_entry(sched->kss_rx_conns.next,
1656 ksock_conn_t, ksnc_rx_list);
1657 list_del(&conn->ksnc_rx_list);
1659 LASSERT(conn->ksnc_rx_scheduled);
1660 LASSERT(conn->ksnc_rx_ready);
1662 /* clear rx_ready in case receive isn't complete.
1663 * Do it BEFORE we call process_recv, since
1664 * data_ready can set it any time after we release
1666 conn->ksnc_rx_ready = 0;
1667 spin_unlock_irqrestore(&sched->kss_lock, flags);
1669 rc = ksocknal_process_receive(conn);
1671 spin_lock_irqsave(&sched->kss_lock, flags);
1673 /* I'm the only one that can clear this flag */
1674 LASSERT(conn->ksnc_rx_scheduled);
1676 /* Did process_receive get everything it wanted? */
1678 conn->ksnc_rx_ready = 1;
1680 if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
1681 conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
1682 /* Conn blocked for a forwarding buffer.
1683 * It will get queued for my attention when
1684 * one becomes available (and it might just
1685 * already have been!). Meanwhile my ref
1686 * on it stays put. */
1687 } else if (conn->ksnc_rx_ready) {
1688 /* reschedule for rx */
1689 list_add_tail (&conn->ksnc_rx_list,
1690 &sched->kss_rx_conns);
1692 conn->ksnc_rx_scheduled = 0;
1694 ksocknal_put_conn(conn);
1700 if (!list_empty (&sched->kss_tx_conns)) {
1701 conn = list_entry(sched->kss_tx_conns.next,
1702 ksock_conn_t, ksnc_tx_list);
1703 list_del (&conn->ksnc_tx_list);
1705 LASSERT(conn->ksnc_tx_scheduled);
1706 LASSERT(conn->ksnc_tx_ready);
1707 LASSERT(!list_empty(&conn->ksnc_tx_queue));
1709 tx = list_entry(conn->ksnc_tx_queue.next,
1710 ksock_tx_t, tx_list);
1711 /* dequeue now so empty list => more to send */
1712 list_del(&tx->tx_list);
1714 /* Clear tx_ready in case send isn't complete. Do
1715 * it BEFORE we call process_transmit, since
1716 * write_space can set it any time after we release
1718 conn->ksnc_tx_ready = 0;
1719 spin_unlock_irqrestore (&sched->kss_lock, flags);
1721 rc = ksocknal_process_transmit(conn, tx);
1723 spin_lock_irqsave (&sched->kss_lock, flags);
1725 if (rc == -ENOMEM || rc == -EAGAIN) {
1726 /* Incomplete send: replace tx on HEAD of tx_queue */
1727 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
1729 /* Complete send; assume space for more */
1730 conn->ksnc_tx_ready = 1;
1733 if (rc == -ENOMEM) {
1734 /* Do nothing; after a short timeout, this
1735 * conn will be reposted on kss_tx_conns. */
1736 } else if (conn->ksnc_tx_ready &&
1737 !list_empty (&conn->ksnc_tx_queue)) {
1738 /* reschedule for tx */
1739 list_add_tail (&conn->ksnc_tx_list,
1740 &sched->kss_tx_conns);
1742 conn->ksnc_tx_scheduled = 0;
1744 ksocknal_put_conn (conn);
1750 if (!list_empty (&sched->kss_zctxdone_list)) {
1752 list_entry(sched->kss_zctxdone_list.next,
1753 ksock_tx_t, tx_list);
1756 list_del (&tx->tx_list);
1757 spin_unlock_irqrestore (&sched->kss_lock, flags);
1759 ksocknal_tx_done (tx, 1);
1761 spin_lock_irqsave (&sched->kss_lock, flags);
1764 if (!did_something || /* nothing to do */
1765 ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1766 spin_unlock_irqrestore (&sched->kss_lock, flags);
1770 if (!did_something) { /* wait for something to do */
1771 rc = wait_event_interruptible (sched->kss_waitq,
1772 !ksocknal_sched_cansleep(sched));
1777 spin_lock_irqsave (&sched->kss_lock, flags);
1781 spin_unlock_irqrestore (&sched->kss_lock, flags);
1782 ksocknal_thread_fini ();
1787 ksocknal_data_ready (struct sock *sk, int n)
1789 unsigned long flags;
1791 ksock_sched_t *sched;
1794 /* interleave correctly with closing sockets... */
1795 read_lock (&ksocknal_data.ksnd_global_lock);
1797 conn = sk->sk_user_data;
1798 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1799 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1800 sk->sk_data_ready (sk, n);
1802 sched = conn->ksnc_scheduler;
1804 spin_lock_irqsave (&sched->kss_lock, flags);
1806 conn->ksnc_rx_ready = 1;
1808 if (!conn->ksnc_rx_scheduled) { /* not being progressed */
1809 list_add_tail(&conn->ksnc_rx_list,
1810 &sched->kss_rx_conns);
1811 conn->ksnc_rx_scheduled = 1;
1812 /* extra ref for scheduler */
1813 atomic_inc (&conn->ksnc_refcount);
1815 wake_up (&sched->kss_waitq);
1818 spin_unlock_irqrestore (&sched->kss_lock, flags);
1821 read_unlock (&ksocknal_data.ksnd_global_lock);
1827 ksocknal_write_space (struct sock *sk)
1829 unsigned long flags;
1831 ksock_sched_t *sched;
1833 /* interleave correctly with closing sockets... */
1834 read_lock (&ksocknal_data.ksnd_global_lock);
1836 conn = sk->sk_user_data;
1838 CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1839 sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1840 (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1841 " ready" : " blocked"),
1842 (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1843 " scheduled" : " idle"),
1844 (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1845 " empty" : " queued"));
1847 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1848 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1849 sk->sk_write_space (sk);
1851 read_unlock (&ksocknal_data.ksnd_global_lock);
1855 if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1856 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1858 sched = conn->ksnc_scheduler;
1860 spin_lock_irqsave (&sched->kss_lock, flags);
1862 conn->ksnc_tx_ready = 1;
1864 if (!conn->ksnc_tx_scheduled && // not being progressed
1865 !list_empty(&conn->ksnc_tx_queue)){//packets to send
1866 list_add_tail (&conn->ksnc_tx_list,
1867 &sched->kss_tx_conns);
1868 conn->ksnc_tx_scheduled = 1;
1869 /* extra ref for scheduler */
1870 atomic_inc (&conn->ksnc_refcount);
1872 wake_up (&sched->kss_waitq);
1875 spin_unlock_irqrestore (&sched->kss_lock, flags);
1878 read_unlock (&ksocknal_data.ksnd_global_lock);
1882 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1885 mm_segment_t oldmm = get_fs();
1888 struct iovec iov = {
1892 struct msghdr msg = {
1897 .msg_control = NULL,
1898 .msg_controllen = 0,
1903 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1910 CERROR ("Unexpected zero rc\n");
1911 return (-ECONNABORTED);
1914 buffer = ((char *)buffer) + rc;
1922 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1925 mm_segment_t oldmm = get_fs();
1928 struct iovec iov = {
1932 struct msghdr msg = {
1937 .msg_control = NULL,
1938 .msg_controllen = 0,
1943 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
1950 return (-ECONNABORTED);
1952 buffer = ((char *)buffer) + rc;
1960 ksocknal_send_hello (ksock_conn_t *conn, __u32 *ipaddrs, int nipaddrs)
1962 /* CAVEAT EMPTOR: this byte flips 'ipaddrs' */
1963 struct socket *sock = conn->ksnc_sock;
1965 ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
1969 LASSERT (conn->ksnc_type != SOCKNAL_CONN_NONE);
1970 LASSERT (nipaddrs <= SOCKNAL_MAX_INTERFACES);
1972 /* No need for getconnsock/putconnsock */
1973 LASSERT (!conn->ksnc_closing);
1975 LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
1976 hmv->magic = cpu_to_le32 (PORTALS_PROTO_MAGIC);
1977 hmv->version_major = cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR);
1978 hmv->version_minor = cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR);
1980 hdr.src_nid = cpu_to_le64 (ksocknal_lib.libnal_ni.ni_pid.nid);
1981 hdr.type = cpu_to_le32 (PTL_MSG_HELLO);
1982 hdr.payload_length = cpu_to_le32 (nipaddrs * sizeof(*ipaddrs));
1984 hdr.msg.hello.type = cpu_to_le32 (conn->ksnc_type);
1985 hdr.msg.hello.incarnation =
1986 cpu_to_le64 (ksocknal_data.ksnd_incarnation);
1988 /* Receiver is eager */
1989 rc = ksocknal_sock_write (sock, &hdr, sizeof(hdr));
1991 CERROR ("Error %d sending HELLO hdr to %u.%u.%u.%u/%d\n",
1992 rc, HIPQUAD(conn->ksnc_ipaddr), conn->ksnc_port);
1999 for (i = 0; i < nipaddrs; i++) {
2000 ipaddrs[i] = __cpu_to_le32 (ipaddrs[i]);
2003 rc = ksocknal_sock_write (sock, ipaddrs, nipaddrs * sizeof(*ipaddrs));
2005 CERROR ("Error %d sending HELLO payload (%d)"
2006 " to %u.%u.%u.%u/%d\n", rc, nipaddrs,
2007 HIPQUAD(conn->ksnc_ipaddr), conn->ksnc_port);
2012 ksocknal_invert_type(int type)
2016 case SOCKNAL_CONN_ANY:
2017 case SOCKNAL_CONN_CONTROL:
2019 case SOCKNAL_CONN_BULK_IN:
2020 return SOCKNAL_CONN_BULK_OUT;
2021 case SOCKNAL_CONN_BULK_OUT:
2022 return SOCKNAL_CONN_BULK_IN;
2024 return (SOCKNAL_CONN_NONE);
2029 ksocknal_recv_hello (ksock_conn_t *conn, ptl_nid_t *nid,
2030 __u64 *incarnation, __u32 *ipaddrs)
2032 struct socket *sock = conn->ksnc_sock;
2038 ptl_magicversion_t *hmv;
2040 hmv = (ptl_magicversion_t *)&hdr.dest_nid;
2041 LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
2043 rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
2045 CERROR ("Error %d reading HELLO from %u.%u.%u.%u\n",
2046 rc, HIPQUAD(conn->ksnc_ipaddr));
2050 if (hmv->magic != le32_to_cpu (PORTALS_PROTO_MAGIC)) {
2051 CERROR ("Bad magic %#08x (%#08x expected) from %u.%u.%u.%u\n",
2052 __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC,
2053 HIPQUAD(conn->ksnc_ipaddr));
2057 if (hmv->version_major != cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
2058 hmv->version_minor != cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
2059 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
2060 " from %u.%u.%u.%u\n",
2061 le16_to_cpu (hmv->version_major),
2062 le16_to_cpu (hmv->version_minor),
2063 PORTALS_PROTO_VERSION_MAJOR,
2064 PORTALS_PROTO_VERSION_MINOR,
2065 HIPQUAD(conn->ksnc_ipaddr));
2069 #if (PORTALS_PROTO_VERSION_MAJOR != 1)
2070 # error "This code only understands protocol version 1.x"
2072 /* version 1 sends magic/version as the dest_nid of a 'hello'
2073 * header, followed by payload full of interface IP addresses.
2074 * Read the rest of it in now... */
2076 rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
2078 CERROR ("Error %d reading rest of HELLO hdr from %u.%u.%u.%u\n",
2079 rc, HIPQUAD(conn->ksnc_ipaddr));
2083 /* ...and check we got what we expected */
2084 if (hdr.type != cpu_to_le32 (PTL_MSG_HELLO)) {
2085 CERROR ("Expecting a HELLO hdr,"
2086 " but got type %d from %u.%u.%u.%u\n",
2087 le32_to_cpu (hdr.type),
2088 HIPQUAD(conn->ksnc_ipaddr));
2092 if (le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
2093 CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY"
2094 "from %u.%u.%u.%u\n", HIPQUAD(conn->ksnc_ipaddr));
2098 if (*nid == PTL_NID_ANY) { /* don't know peer's nid yet */
2099 *nid = le64_to_cpu(hdr.src_nid);
2100 } else if (*nid != le64_to_cpu (hdr.src_nid)) {
2101 CERROR ("Connected to nid "LPX64"@%u.%u.%u.%u "
2102 "but expecting "LPX64"\n",
2103 le64_to_cpu (hdr.src_nid),
2104 HIPQUAD(conn->ksnc_ipaddr), *nid);
2108 type = __le32_to_cpu(hdr.msg.hello.type);
2110 if (conn->ksnc_type == SOCKNAL_CONN_NONE) {
2111 /* I've accepted this connection; peer determines type */
2112 conn->ksnc_type = ksocknal_invert_type(type);
2113 if (conn->ksnc_type == SOCKNAL_CONN_NONE) {
2114 CERROR ("Unexpected type %d from "LPX64"@%u.%u.%u.%u\n",
2115 type, *nid, HIPQUAD(conn->ksnc_ipaddr));
2118 } else if (ksocknal_invert_type(type) != conn->ksnc_type) {
2119 CERROR ("Mismatched types: me %d, "LPX64"@%u.%u.%u.%u %d\n",
2120 conn->ksnc_type, *nid, HIPQUAD(conn->ksnc_ipaddr),
2121 le32_to_cpu(hdr.msg.hello.type));
2125 *incarnation = le64_to_cpu(hdr.msg.hello.incarnation);
2127 nips = __le32_to_cpu (hdr.payload_length) / sizeof (__u32);
2129 if (nips > SOCKNAL_MAX_INTERFACES ||
2130 nips * sizeof(__u32) != __le32_to_cpu (hdr.payload_length)) {
2131 CERROR("Bad payload length %d from "LPX64"@%u.%u.%u.%u\n",
2132 __le32_to_cpu (hdr.payload_length),
2133 *nid, HIPQUAD(conn->ksnc_ipaddr));
2139 rc = ksocknal_sock_read (sock, ipaddrs, nips * sizeof(*ipaddrs));
2141 CERROR ("Error %d reading IPs from "LPX64"@%u.%u.%u.%u\n",
2142 rc, *nid, HIPQUAD(conn->ksnc_ipaddr));
2146 for (i = 0; i < nips; i++) {
2147 ipaddrs[i] = __le32_to_cpu(ipaddrs[i]);
2149 if (ipaddrs[i] == 0) {
2150 CERROR("Zero IP[%d] from "LPX64"@%u.%u.%u.%u\n",
2151 i, *nid, HIPQUAD(conn->ksnc_ipaddr));
2160 ksocknal_get_conn_tunables (ksock_conn_t *conn, int *txmem, int *rxmem, int *nagle)
2162 mm_segment_t oldmm = get_fs ();
2163 struct socket *sock = conn->ksnc_sock;
2167 rc = ksocknal_getconnsock (conn);
2169 LASSERT (conn->ksnc_closing);
2170 *txmem = *rxmem = *nagle = 0;
2171 return (-ESHUTDOWN);
2176 len = sizeof(*txmem);
2177 rc = sock_getsockopt(sock, SOL_SOCKET, SO_SNDBUF,
2178 (char *)txmem, &len);
2180 len = sizeof(*rxmem);
2181 rc = sock_getsockopt(sock, SOL_SOCKET, SO_RCVBUF,
2182 (char *)rxmem, &len);
2185 len = sizeof(*nagle);
2186 rc = sock->ops->getsockopt(sock, SOL_TCP, TCP_NODELAY,
2187 (char *)nagle, &len);
2191 ksocknal_putconnsock (conn);
2196 *txmem = *rxmem = *nagle = 0;
2202 ksocknal_setup_sock (struct socket *sock)
2204 mm_segment_t oldmm = get_fs ();
2211 struct linger linger;
2213 sock->sk->sk_allocation = GFP_NOFS;
2215 /* Ensure this socket aborts active sends immediately when we close
2219 linger.l_linger = 0;
2222 rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2223 (char *)&linger, sizeof (linger));
2226 CERROR ("Can't set SO_LINGER: %d\n", rc);
2232 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2233 (char *)&option, sizeof (option));
2236 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2240 if (!ksocknal_tunables.ksnd_nagle) {
2244 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2245 (char *)&option, sizeof (option));
2248 CERROR ("Can't disable nagle: %d\n", rc);
2253 if (ksocknal_tunables.ksnd_buffer_size > 0) {
2254 option = ksocknal_tunables.ksnd_buffer_size;
2257 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2258 (char *)&option, sizeof (option));
2261 CERROR ("Can't set send buffer %d: %d\n",
2267 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2268 (char *)&option, sizeof (option));
2271 CERROR ("Can't set receive buffer %d: %d\n",
2277 /* snapshot tunables */
2278 keep_idle = ksocknal_tunables.ksnd_keepalive_idle;
2279 keep_count = ksocknal_tunables.ksnd_keepalive_count;
2280 keep_intvl = ksocknal_tunables.ksnd_keepalive_intvl;
2282 do_keepalive = (keep_idle > 0 && keep_count > 0 && keep_intvl > 0);
2284 option = (do_keepalive ? 1 : 0);
2286 rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
2287 (char *)&option, sizeof (option));
2290 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2298 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2299 (char *)&keep_idle, sizeof (keep_idle));
2302 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2307 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2308 (char *)&keep_intvl, sizeof (keep_intvl));
2311 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2316 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2317 (char *)&keep_count, sizeof (keep_count));
2320 CERROR ("Can't set TCP_KEEPCNT: %d\n", rc);
2328 ksocknal_connect_peer (ksock_route_t *route, int type)
2330 struct sockaddr_in ipaddr;
2331 mm_segment_t oldmm = get_fs();
2334 struct socket *sock;
2337 rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2339 CERROR ("Can't create autoconnect socket: %d\n", rc);
2343 /* Ugh; have to map_fd for compatibility with sockets passed in
2344 * from userspace. And we actually need the sock->file refcounting
2345 * that this gives you :) */
2347 fd = sock_map_fd (sock);
2349 sock_release (sock);
2350 CERROR ("sock_map_fd error %d\n", fd);
2354 /* NB the fd now owns the ref on sock->file */
2355 LASSERT (sock->file != NULL);
2356 LASSERT (file_count(sock->file) == 1);
2358 /* Set the socket timeouts, so our connection attempt completes in
2360 tv.tv_sec = ksocknal_tunables.ksnd_io_timeout;
2364 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2365 (char *)&tv, sizeof (tv));
2368 CERROR ("Can't set send timeout %d: %d\n",
2369 ksocknal_tunables.ksnd_io_timeout, rc);
2374 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2375 (char *)&tv, sizeof (tv));
2378 CERROR ("Can't set receive timeout %d: %d\n",
2379 ksocknal_tunables.ksnd_io_timeout, rc);
2383 if (route->ksnr_myipaddr != 0) {
2384 /* Bind to the local IP address */
2385 memset (&ipaddr, 0, sizeof (ipaddr));
2386 ipaddr.sin_family = AF_INET;
2387 ipaddr.sin_port = htons (0); /* ANY */
2388 ipaddr.sin_addr.s_addr = htonl(route->ksnr_myipaddr);
2390 rc = sock->ops->bind (sock, (struct sockaddr *)&ipaddr,
2393 CERROR ("Can't bind to local IP %u.%u.%u.%u: %d\n",
2394 HIPQUAD(route->ksnr_myipaddr), rc);
2399 memset (&ipaddr, 0, sizeof (ipaddr));
2400 ipaddr.sin_family = AF_INET;
2401 ipaddr.sin_port = htons (route->ksnr_port);
2402 ipaddr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2404 rc = sock->ops->connect (sock, (struct sockaddr *)&ipaddr,
2405 sizeof (ipaddr), sock->file->f_flags);
2407 CERROR ("Can't connect to nid "LPX64
2408 " local IP: %u.%u.%u.%u,"
2409 " remote IP: %u.%u.%u.%u/%d: %d\n",
2410 route->ksnr_peer->ksnp_nid,
2411 HIPQUAD(route->ksnr_myipaddr),
2412 HIPQUAD(route->ksnr_ipaddr),
2413 route->ksnr_port, rc);
2417 rc = ksocknal_create_conn (route, sock, type);
2419 /* Take an extra ref on sock->file to compensate for the
2420 * upcoming close which will lose fd's ref on it. */
2421 get_file (sock->file);
2430 ksocknal_autoconnect (ksock_route_t *route)
2432 LIST_HEAD (zombies);
2435 unsigned long flags;
2440 for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
2441 if ((route->ksnr_connecting & (1 << type)) != 0)
2443 LASSERT (type < SOCKNAL_CONN_NTYPES);
2445 rc = ksocknal_connect_peer (route, type);
2450 /* successfully autoconnected: create_conn did the
2451 * route/conn binding and scheduled any blocked packets */
2453 if (route->ksnr_connecting == 0) {
2454 /* No more connections required */
2459 /* Connection attempt failed */
2461 write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2463 peer = route->ksnr_peer;
2464 route->ksnr_connecting = 0;
2466 /* This is a retry rather than a new connection */
2467 LASSERT (route->ksnr_retry_interval != 0);
2468 route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2469 route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2470 SOCKNAL_MAX_RECONNECT_INTERVAL);
2472 if (!list_empty (&peer->ksnp_tx_queue) &&
2473 ksocknal_find_connecting_route_locked (peer) == NULL) {
2474 LASSERT (list_empty (&peer->ksnp_conns));
2476 /* None of the connections that the blocked packets are
2477 * waiting for have been successful. Complete them now... */
2479 tx = list_entry (peer->ksnp_tx_queue.next,
2480 ksock_tx_t, tx_list);
2481 list_del (&tx->tx_list);
2482 list_add_tail (&tx->tx_list, &zombies);
2483 } while (!list_empty (&peer->ksnp_tx_queue));
2486 #if 0 /* irrelevent with only eager routes */
2487 if (!route->ksnr_deleted) {
2488 /* make this route least-favourite for re-selection */
2489 list_del(&route->ksnr_list);
2490 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
2493 write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2495 while (!list_empty (&zombies)) {
2496 char ipbuf[PTL_NALFMT_SIZE];
2497 char ipbuf2[PTL_NALFMT_SIZE];
2498 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2500 CERROR ("Deleting packet type %d len %d ("LPX64" %s->"LPX64" %s)\n",
2501 le32_to_cpu (tx->tx_hdr->type),
2502 le32_to_cpu (tx->tx_hdr->payload_length),
2503 le64_to_cpu (tx->tx_hdr->src_nid),
2504 portals_nid2str(SOCKNAL,
2505 le64_to_cpu(tx->tx_hdr->src_nid),
2507 le64_to_cpu (tx->tx_hdr->dest_nid),
2508 portals_nid2str(SOCKNAL,
2509 le64_to_cpu(tx->tx_hdr->src_nid),
2512 list_del (&tx->tx_list);
2514 ksocknal_tx_done (tx, 0);
2519 ksocknal_autoconnectd (void *arg)
2521 long id = (long)arg;
2523 unsigned long flags;
2524 ksock_route_t *route;
2527 snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2528 kportal_daemonize (name);
2529 kportal_blockallsigs ();
2531 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2533 while (!ksocknal_data.ksnd_shuttingdown) {
2535 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2536 route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2537 ksock_route_t, ksnr_connect_list);
2539 list_del (&route->ksnr_connect_list);
2540 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2542 ksocknal_autoconnect (route);
2543 ksocknal_put_route (route);
2545 spin_lock_irqsave(&ksocknal_data.ksnd_autoconnectd_lock,
2550 spin_unlock_irqrestore(&ksocknal_data.ksnd_autoconnectd_lock,
2553 rc = wait_event_interruptible(ksocknal_data.ksnd_autoconnectd_waitq,
2554 ksocknal_data.ksnd_shuttingdown ||
2555 !list_empty(&ksocknal_data.ksnd_autoconnectd_routes));
2557 spin_lock_irqsave(&ksocknal_data.ksnd_autoconnectd_lock, flags);
2560 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2562 ksocknal_thread_fini ();
2567 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2569 /* We're called with a shared lock on ksnd_global_lock */
2571 struct list_head *ctmp;
2573 list_for_each (ctmp, &peer->ksnp_conns) {
2574 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2576 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2577 LASSERT (!conn->ksnc_closing);
2579 if (conn->ksnc_sock->sk->sk_err != 0) {
2580 /* Something (e.g. failed keepalive) set the socket error */
2581 atomic_inc (&conn->ksnc_refcount);
2582 CERROR ("Socket error %d: "LPX64" %p %d.%d.%d.%d\n",
2583 conn->ksnc_sock->sk->sk_err, peer->ksnp_nid,
2584 conn, HIPQUAD(conn->ksnc_ipaddr));
2588 if (conn->ksnc_rx_started &&
2589 time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2590 /* Timed out incomplete incoming message */
2591 atomic_inc (&conn->ksnc_refcount);
2592 CERROR ("Timed out RX from "LPX64" %p %d.%d.%d.%d\n",
2593 peer->ksnp_nid,conn,HIPQUAD(conn->ksnc_ipaddr));
2597 if ((!list_empty (&conn->ksnc_tx_queue) ||
2598 conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2599 time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2600 /* Timed out messages queued for sending or
2601 * buffered in the socket's send buffer */
2602 atomic_inc (&conn->ksnc_refcount);
2603 CERROR ("Timed out TX to "LPX64" %s%d %p %d.%d.%d.%d\n",
2605 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2606 conn->ksnc_sock->sk->sk_wmem_queued, conn,
2607 HIPQUAD(conn->ksnc_ipaddr));
2616 ksocknal_check_peer_timeouts (int idx)
2618 struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2619 struct list_head *ptmp;
2624 /* NB. We expect to have a look at all the peers and not find any
2625 * connections to time out, so we just use a shared lock while we
2627 read_lock (&ksocknal_data.ksnd_global_lock);
2629 list_for_each (ptmp, peers) {
2630 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2631 conn = ksocknal_find_timed_out_conn (peer);
2634 read_unlock (&ksocknal_data.ksnd_global_lock);
2636 CERROR ("Timeout out conn->"LPX64" ip %d.%d.%d.%d:%d\n",
2638 HIPQUAD(conn->ksnc_ipaddr),
2640 ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2642 /* NB we won't find this one again, but we can't
2643 * just proceed with the next peer, since we dropped
2644 * ksnd_global_lock and it might be dead already! */
2645 ksocknal_put_conn (conn);
2650 read_unlock (&ksocknal_data.ksnd_global_lock);
2654 ksocknal_reaper (void *arg)
2657 unsigned long flags;
2659 ksock_sched_t *sched;
2660 struct list_head enomem_conns;
2665 unsigned long deadline = jiffies;
2667 kportal_daemonize ("ksocknal_reaper");
2668 kportal_blockallsigs ();
2670 INIT_LIST_HEAD(&enomem_conns);
2671 init_waitqueue_entry (&wait, current);
2673 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2675 while (!ksocknal_data.ksnd_shuttingdown) {
2677 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2678 conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2679 ksock_conn_t, ksnc_list);
2680 list_del (&conn->ksnc_list);
2682 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2684 ksocknal_terminate_conn (conn);
2685 ksocknal_put_conn (conn);
2687 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2691 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2692 conn = list_entry (ksocknal_data.ksnd_zombie_conns.next,
2693 ksock_conn_t, ksnc_list);
2694 list_del (&conn->ksnc_list);
2696 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2698 ksocknal_destroy_conn (conn);
2700 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2704 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2705 list_add(&enomem_conns, &ksocknal_data.ksnd_enomem_conns);
2706 list_del_init(&ksocknal_data.ksnd_enomem_conns);
2709 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2711 /* reschedule all the connections that stalled with ENOMEM... */
2713 while (!list_empty (&enomem_conns)) {
2714 conn = list_entry (enomem_conns.next,
2715 ksock_conn_t, ksnc_tx_list);
2716 list_del (&conn->ksnc_tx_list);
2718 sched = conn->ksnc_scheduler;
2720 spin_lock_irqsave (&sched->kss_lock, flags);
2722 LASSERT (conn->ksnc_tx_scheduled);
2723 conn->ksnc_tx_ready = 1;
2724 list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
2725 wake_up (&sched->kss_waitq);
2727 spin_unlock_irqrestore (&sched->kss_lock, flags);
2731 /* careful with the jiffy wrap... */
2732 while ((timeout = (int)(deadline - jiffies)) <= 0) {
2735 int chunk = ksocknal_data.ksnd_peer_hash_size;
2737 /* Time to check for timeouts on a few more peers: I do
2738 * checks every 'p' seconds on a proportion of the peer
2739 * table and I need to check every connection 'n' times
2740 * within a timeout interval, to ensure I detect a
2741 * timeout on any connection within (n+1)/n times the
2742 * timeout interval. */
2744 if (ksocknal_tunables.ksnd_io_timeout > n * p)
2745 chunk = (chunk * n * p) /
2746 ksocknal_tunables.ksnd_io_timeout;
2750 for (i = 0; i < chunk; i++) {
2751 ksocknal_check_peer_timeouts (peer_index);
2752 peer_index = (peer_index + 1) %
2753 ksocknal_data.ksnd_peer_hash_size;
2759 if (nenomem_conns != 0) {
2760 /* Reduce my timeout if I rescheduled ENOMEM conns.
2761 * This also prevents me getting woken immediately
2762 * if any go back on my enomem list. */
2763 timeout = SOCKNAL_ENOMEM_RETRY;
2765 ksocknal_data.ksnd_reaper_waketime = jiffies + timeout;
2767 set_current_state (TASK_INTERRUPTIBLE);
2768 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2770 if (!ksocknal_data.ksnd_shuttingdown &&
2771 list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2772 list_empty (&ksocknal_data.ksnd_zombie_conns))
2773 schedule_timeout (timeout);
2775 set_current_state (TASK_RUNNING);
2776 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2778 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2781 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2783 ksocknal_thread_fini ();
2787 lib_nal_t ksocknal_lib = {
2788 libnal_data: &ksocknal_data, /* NAL private data */
2789 libnal_send: ksocknal_send,
2790 libnal_send_pages: ksocknal_send_pages,
2791 libnal_recv: ksocknal_recv,
2792 libnal_recv_pages: ksocknal_recv_pages,
2793 libnal_dist: ksocknal_dist