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 /* OK to ignore 'flags'; we're only ever serialise threads and
94 * never need to lock out interrupts */
95 spin_lock(&data->ksnd_nal_cb_lock);
99 ksocknal_sti(nal_cb_t *nal, unsigned long *flags)
101 ksock_nal_data_t *data;
102 data = nal->nal_data;
104 /* OK to ignore 'flags'; we're only ever serialise threads and
105 * never need to lock out interrupts */
106 spin_unlock(&data->ksnd_nal_cb_lock);
110 ksocknal_callback(nal_cb_t *nal, void *private, lib_eq_t *eq, ptl_event_t *ev)
112 /* holding ksnd_nal_cb_lock */
114 if (eq->event_callback != NULL)
115 eq->event_callback(ev);
117 if (waitqueue_active(&ksocknal_data.ksnd_yield_waitq))
118 wake_up_all(&ksocknal_data.ksnd_yield_waitq);
122 ksocknal_dist(nal_cb_t *nal, ptl_nid_t nid, unsigned long *dist)
124 /* I would guess that if ksocknal_get_peer (nid) == NULL,
125 and we're not routing, then 'nid' is very distant :) */
126 if ( nal->ni.nid == nid ) {
136 ksocknal_free_ltx (ksock_ltx_t *ltx)
138 atomic_dec(&ksocknal_data.ksnd_nactive_ltxs);
139 PORTAL_FREE(ltx, ltx->ltx_desc_size);
142 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
144 ksocknal_kvaddr_to_page (unsigned long vaddr)
148 if (vaddr >= VMALLOC_START &&
150 page = vmalloc_to_page ((void *)vaddr);
152 else if (vaddr >= PKMAP_BASE &&
153 vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
154 page = vmalloc_to_page ((void *)vaddr);
155 /* in 2.4 ^ just walks the page tables */
158 page = virt_to_page (vaddr);
169 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
171 struct socket *sock = conn->ksnc_sock;
172 struct iovec *iov = tx->tx_iov;
173 int fragsize = iov->iov_len;
174 unsigned long vaddr = (unsigned long)iov->iov_base;
175 int more = (tx->tx_niov > 1) ||
176 (tx->tx_nkiov > 0) ||
177 (!list_empty (&conn->ksnc_tx_queue));
178 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
179 int offset = vaddr & (PAGE_SIZE - 1);
180 int zcsize = MIN (fragsize, PAGE_SIZE - offset);
185 /* NB we can't trust socket ops to either consume our iovs
186 * or leave them alone, so we only send 1 frag at a time. */
187 LASSERT (fragsize <= tx->tx_resid);
188 LASSERT (tx->tx_niov > 0);
190 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
191 if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
192 (sock->sk->route_caps & NETIF_F_SG) &&
193 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
194 (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
196 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
197 (void *)vaddr, page, page_address(page), offset, zcsize);
199 if (fragsize > zcsize) {
204 rc = tcp_sendpage_zccd(sock, page, offset, zcsize,
205 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
210 /* NB don't pass tx's iov; sendmsg may or may not update it */
211 struct iovec fragiov = { .iov_base = (void *)vaddr,
212 .iov_len = fragsize};
213 struct msghdr msg = {
220 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
222 mm_segment_t oldmm = get_fs();
225 rc = sock_sendmsg(sock, &msg, fragsize);
232 if (rc < iov->iov_len) {
233 /* didn't send whole iov entry... */
234 iov->iov_base = (void *)(vaddr + rc);
246 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
248 struct socket *sock = conn->ksnc_sock;
249 ptl_kiov_t *kiov = tx->tx_kiov;
250 int fragsize = kiov->kiov_len;
251 struct page *page = kiov->kiov_page;
252 int offset = kiov->kiov_offset;
253 int more = (tx->tx_nkiov > 1) ||
254 (!list_empty (&conn->ksnc_tx_queue));
257 /* NB we can't trust socket ops to either consume our iovs
258 * or leave them alone, so we only send 1 frag at a time. */
259 LASSERT (fragsize <= tx->tx_resid);
260 LASSERT (offset + fragsize <= PAGE_SIZE);
261 LASSERT (tx->tx_niov == 0);
262 LASSERT (tx->tx_nkiov > 0);
265 if (fragsize >= ksocknal_tunables.ksnd_zc_min_frag &&
266 (sock->sk->route_caps & NETIF_F_SG) &&
267 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
269 CDEBUG(D_NET, "page %p + offset %x for %d\n",
270 page, offset, fragsize);
272 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
273 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
278 char *addr = ((char *)kmap (page)) + offset;
279 struct iovec fragiov = {.iov_base = addr,
280 .iov_len = fragsize};
281 struct msghdr msg = {
288 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
290 mm_segment_t oldmm = get_fs();
293 rc = sock_sendmsg(sock, &msg, fragsize);
303 kiov->kiov_offset = offset + rc;
304 kiov->kiov_len = fragsize - rc;
315 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
319 if (ksocknal_data.ksnd_stall_tx != 0) {
320 set_current_state (TASK_UNINTERRUPTIBLE);
321 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
324 LASSERT (tx->tx_resid != 0);
326 rc = ksocknal_getconnsock (conn);
328 LASSERT (conn->ksnc_closing);
333 if (ksocknal_data.ksnd_enomem_tx > 0) {
335 ksocknal_data.ksnd_enomem_tx--;
337 } else if (tx->tx_niov != 0) {
338 rc = ksocknal_send_iov (conn, tx);
340 rc = ksocknal_send_kiov (conn, tx);
344 /* Didn't write anything.
346 * NB: rc == 0 and rc == -EAGAIN both mean try
347 * again later (linux stack returns -EAGAIN for
348 * this, but Adaptech TOE returns 0).
350 * Also, sends never fail with -ENOMEM, just
351 * -EAGAIN, but with the added bonus that we can't
352 * expect write_space() to call us back to tell us
353 * when to try sending again. We use the
354 * SOCK_NOSPACE flag to diagnose... */
356 LASSERT(rc != -ENOMEM);
358 if (rc == 0 || rc == -EAGAIN) {
359 if (test_bit(SOCK_NOSPACE,
360 &conn->ksnc_sock->flags)) {
366 if ((counter & (-counter)) == counter)
367 CWARN("%d ENOMEM tx %p\n",
377 /* Consider the connection alive since we managed to chuck
378 * more data into it. Really, we'd like to consider it
379 * alive only when the peer ACKs something, but
380 * write_space() only gets called back while SOCK_NOSPACE
381 * is set. Instead, we presume peer death has occurred if
382 * the socket doesn't drain within a timout */
383 conn->ksnc_tx_deadline = jiffies +
384 ksocknal_tunables.ksnd_io_timeout * HZ;
385 conn->ksnc_peer->ksnp_last_alive = jiffies;
387 } while (tx->tx_resid != 0);
389 ksocknal_putconnsock (conn);
394 ksocknal_eager_ack (ksock_conn_t *conn)
397 mm_segment_t oldmm = get_fs();
398 struct socket *sock = conn->ksnc_sock;
400 /* Remind the socket to ACK eagerly. If I don't, the socket might
401 * think I'm about to send something it could piggy-back the ACK
402 * on, introducing delay in completing zero-copy sends in my
406 sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
407 (char *)&opt, sizeof (opt));
412 ksocknal_recv_iov (ksock_conn_t *conn)
414 struct iovec *iov = conn->ksnc_rx_iov;
415 int fragsize = iov->iov_len;
416 unsigned long vaddr = (unsigned long)iov->iov_base;
417 struct iovec fragiov = { .iov_base = (void *)vaddr,
418 .iov_len = fragsize};
419 struct msghdr msg = {
428 mm_segment_t oldmm = get_fs();
431 /* NB we can't trust socket ops to either consume our iovs
432 * or leave them alone, so we only receive 1 frag at a time. */
433 LASSERT (conn->ksnc_rx_niov > 0);
434 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
437 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
438 /* NB this is just a boolean............................^ */
444 /* received something... */
445 conn->ksnc_peer->ksnp_last_alive = jiffies;
446 conn->ksnc_rx_deadline = jiffies +
447 ksocknal_tunables.ksnd_io_timeout * HZ;
448 mb(); /* order with setting rx_started */
449 conn->ksnc_rx_started = 1;
451 conn->ksnc_rx_nob_wanted -= rc;
452 conn->ksnc_rx_nob_left -= rc;
455 iov->iov_base = (void *)(vaddr + rc);
456 iov->iov_len = fragsize - rc;
461 conn->ksnc_rx_niov--;
466 ksocknal_recv_kiov (ksock_conn_t *conn)
468 ptl_kiov_t *kiov = conn->ksnc_rx_kiov;
469 struct page *page = kiov->kiov_page;
470 int offset = kiov->kiov_offset;
471 int fragsize = kiov->kiov_len;
472 unsigned long vaddr = ((unsigned long)kmap (page)) + offset;
473 struct iovec fragiov = { .iov_base = (void *)vaddr,
474 .iov_len = fragsize};
475 struct msghdr msg = {
484 mm_segment_t oldmm = get_fs();
487 /* NB we can't trust socket ops to either consume our iovs
488 * or leave them alone, so we only receive 1 frag at a time. */
489 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
490 LASSERT (conn->ksnc_rx_nkiov > 0);
491 LASSERT (offset + fragsize <= PAGE_SIZE);
494 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
495 /* NB this is just a boolean............................^ */
503 /* received something... */
504 conn->ksnc_peer->ksnp_last_alive = jiffies;
505 conn->ksnc_rx_deadline = jiffies +
506 ksocknal_tunables.ksnd_io_timeout * HZ;
507 mb(); /* order with setting rx_started */
508 conn->ksnc_rx_started = 1;
510 conn->ksnc_rx_nob_wanted -= rc;
511 conn->ksnc_rx_nob_left -= rc;
514 kiov->kiov_offset = offset + rc;
515 kiov->kiov_len = fragsize - rc;
519 conn->ksnc_rx_kiov++;
520 conn->ksnc_rx_nkiov--;
525 ksocknal_receive (ksock_conn_t *conn)
527 /* Return 1 on success, 0 on EOF, < 0 on error.
528 * Caller checks ksnc_rx_nob_wanted to determine
529 * progress/completion. */
533 if (ksocknal_data.ksnd_stall_rx != 0) {
534 set_current_state (TASK_UNINTERRUPTIBLE);
535 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
538 rc = ksocknal_getconnsock (conn);
540 LASSERT (conn->ksnc_closing);
545 if (conn->ksnc_rx_niov != 0)
546 rc = ksocknal_recv_iov (conn);
548 rc = ksocknal_recv_kiov (conn);
551 /* error/EOF or partial receive */
554 } else if (rc == 0 && conn->ksnc_rx_started) {
555 /* EOF in the middle of a message */
561 /* Completed a fragment */
563 if (conn->ksnc_rx_nob_wanted == 0) {
564 /* Completed a message segment (header or payload) */
565 if ((ksocknal_tunables.ksnd_eager_ack & conn->ksnc_type) != 0 &&
566 (conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
567 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
568 /* Remind the socket to ack eagerly... */
569 ksocknal_eager_ack(conn);
576 ksocknal_putconnsock (conn);
582 ksocknal_zc_callback (zccd_t *zcd)
584 ksock_tx_t *tx = KSOCK_ZCCD_2_TX(zcd);
585 ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
589 /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
591 spin_lock_irqsave (&sched->kss_lock, flags);
593 list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
594 wake_up (&sched->kss_waitq);
596 spin_unlock_irqrestore (&sched->kss_lock, flags);
602 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
607 if (tx->tx_conn != NULL) {
608 /* This tx got queued on a conn; do the accounting... */
609 atomic_sub (tx->tx_nob, &tx->tx_conn->ksnc_tx_nob);
611 /* zero copy completion isn't always from
612 * process_transmit() so it needs to keep a ref on
615 ksocknal_put_conn (tx->tx_conn);
621 if (tx->tx_isfwd) { /* was a forwarded packet? */
622 kpr_fwd_done (&ksocknal_data.ksnd_router,
623 KSOCK_TX_2_KPR_FWD_DESC (tx),
624 (tx->tx_resid == 0) ? 0 : -ECONNABORTED);
630 ltx = KSOCK_TX_2_KSOCK_LTX (tx);
632 lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie,
633 (tx->tx_resid == 0) ? PTL_OK : PTL_FAIL);
635 ksocknal_free_ltx (ltx);
640 ksocknal_tx_launched (ksock_tx_t *tx)
643 if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
644 ksock_conn_t *conn = tx->tx_conn;
646 /* zccd skbufs are still in-flight. First take a ref on
647 * conn, so it hangs about for ksocknal_tx_done... */
648 atomic_inc (&conn->ksnc_refcount);
650 /* ...then drop the initial ref on zccd, so the zero copy
651 * callback can occur */
652 zccd_put (&tx->tx_zccd);
656 /* Any zero-copy-ness (if any) has completed; I can complete the
657 * transmit now, avoiding an extra schedule */
658 ksocknal_tx_done (tx, 0);
662 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
667 rc = ksocknal_transmit (conn, tx);
669 CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
671 if (tx->tx_resid == 0) {
672 /* Sent everything OK */
675 ksocknal_tx_launched (tx);
683 /* Queue on ksnd_enomem_conns for retry after a timeout */
684 spin_lock_irqsave(&ksocknal_data.ksnd_reaper_lock, flags);
686 /* enomem list takes over scheduler's ref... */
687 LASSERT (conn->ksnc_tx_scheduled);
688 list_add_tail(&conn->ksnc_tx_list,
689 &ksocknal_data.ksnd_enomem_conns);
690 if (!time_after_eq(jiffies + SOCKNAL_ENOMEM_RETRY,
691 ksocknal_data.ksnd_reaper_waketime))
692 wake_up (&ksocknal_data.ksnd_reaper_waitq);
694 spin_unlock_irqrestore(&ksocknal_data.ksnd_reaper_lock, flags);
701 if (!conn->ksnc_closing)
702 CERROR("[%p] Error %d on write to "LPX64
703 " ip %d.%d.%d.%d:%d\n", conn, rc,
704 conn->ksnc_peer->ksnp_nid,
705 HIPQUAD(conn->ksnc_ipaddr),
708 ksocknal_close_conn_and_siblings (conn, rc);
709 ksocknal_tx_launched (tx);
715 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
719 /* called holding write lock on ksnd_global_lock */
721 LASSERT (!route->ksnr_deleted);
722 LASSERT ((route->ksnr_connected & (1 << SOCKNAL_CONN_ANY)) == 0);
723 LASSERT ((route->ksnr_connected & KSNR_TYPED_ROUTES) != KSNR_TYPED_ROUTES);
724 LASSERT (!route->ksnr_connecting);
726 if (ksocknal_tunables.ksnd_typed_conns)
727 route->ksnr_connecting =
728 KSNR_TYPED_ROUTES & ~route->ksnr_connected;
730 route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
732 atomic_inc (&route->ksnr_refcount); /* extra ref for asynchd */
734 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
736 list_add_tail (&route->ksnr_connect_list,
737 &ksocknal_data.ksnd_autoconnectd_routes);
738 wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
740 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
744 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
746 char ipbuf[PTL_NALFMT_SIZE];
747 ptl_nid_t target_nid;
749 ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
755 CERROR ("Can't send packet to "LPX64
756 " %s: routed target is not a peer\n",
757 nid, portals_nid2str(SOCKNAL, nid, ipbuf));
761 rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
764 CERROR ("Can't route to "LPX64" %s: router error %d\n",
765 nid, portals_nid2str(SOCKNAL, nid, ipbuf), rc);
769 peer = ksocknal_find_peer_locked (target_nid);
773 CERROR ("Can't send packet to "LPX64" %s: no peer entry\n",
774 target_nid, portals_nid2str(SOCKNAL, target_nid, ipbuf));
779 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
781 struct list_head *tmp;
782 ksock_conn_t *typed = NULL;
784 ksock_conn_t *fallback = NULL;
787 /* Find the conn with the shortest tx queue */
788 list_for_each (tmp, &peer->ksnp_conns) {
789 ksock_conn_t *c = list_entry(tmp, ksock_conn_t, ksnc_list);
790 int nob = atomic_read(&c->ksnc_tx_nob) +
791 c->ksnc_sock->sk->sk_wmem_queued;
793 LASSERT (!c->ksnc_closing);
795 if (fallback == NULL || nob < fnob) {
800 if (!ksocknal_tunables.ksnd_typed_conns)
803 switch (c->ksnc_type) {
806 case SOCKNAL_CONN_ANY:
808 case SOCKNAL_CONN_BULK_IN:
810 case SOCKNAL_CONN_BULK_OUT:
811 if (tx->tx_nob < ksocknal_tunables.ksnd_min_bulk)
814 case SOCKNAL_CONN_CONTROL:
815 if (tx->tx_nob >= ksocknal_tunables.ksnd_min_bulk)
820 if (typed == NULL || nob < tnob) {
826 /* prefer the typed selection */
827 return ((typed != NULL) ? typed : fallback);
831 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
834 ksock_sched_t *sched = conn->ksnc_scheduler;
836 /* called holding global lock (read or irq-write) and caller may
837 * not have dropped this lock between finding conn and calling me,
838 * so we don't need the {get,put}connsock dance to deref
840 LASSERT(!conn->ksnc_closing);
841 LASSERT(tx->tx_resid == tx->tx_nob);
843 CDEBUG (D_NET, "Sending to "LPX64" ip %d.%d.%d.%d:%d\n",
844 conn->ksnc_peer->ksnp_nid,
845 HIPQUAD(conn->ksnc_ipaddr),
848 atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
852 zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
853 /* NB this sets 1 ref on zccd, so the callback can only occur after
854 * I've released this ref. */
856 spin_lock_irqsave (&sched->kss_lock, flags);
858 conn->ksnc_tx_deadline = jiffies +
859 ksocknal_tunables.ksnd_io_timeout * HZ;
860 mb(); /* order with list_add_tail */
862 list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
864 if (conn->ksnc_tx_ready && /* able to send */
865 !conn->ksnc_tx_scheduled) { /* not scheduled to send */
866 /* +1 ref for scheduler */
867 atomic_inc (&conn->ksnc_refcount);
868 list_add_tail (&conn->ksnc_tx_list,
869 &sched->kss_tx_conns);
870 conn->ksnc_tx_scheduled = 1;
871 wake_up (&sched->kss_waitq);
874 spin_unlock_irqrestore (&sched->kss_lock, flags);
878 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
880 struct list_head *tmp;
881 ksock_route_t *route;
882 ksock_route_t *candidate = NULL;
886 list_for_each (tmp, &peer->ksnp_routes) {
887 route = list_entry (tmp, ksock_route_t, ksnr_list);
888 bits = route->ksnr_connected;
890 if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES ||
891 (bits & (1 << SOCKNAL_CONN_ANY)) != 0 ||
892 route->ksnr_connecting != 0) {
893 /* All typed connections have been established, or
894 * an untyped connection has been established, or
895 * connections are currently being established */
900 /* too soon to retry this guy? */
901 if (!time_after_eq (jiffies, route->ksnr_timeout))
904 /* always do eager routes */
905 if (route->ksnr_eager)
908 if (candidate == NULL) {
909 /* If we don't find any other route that is fully
910 * connected or connecting, the first connectable
911 * route is returned. If it fails to connect, it
912 * will get placed at the end of the list */
917 return (found ? NULL : candidate);
921 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
923 struct list_head *tmp;
924 ksock_route_t *route;
926 list_for_each (tmp, &peer->ksnp_routes) {
927 route = list_entry (tmp, ksock_route_t, ksnr_list);
929 if (route->ksnr_connecting != 0)
937 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
942 ksock_route_t *route;
945 /* Ensure the frags we've been given EXACTLY match the number of
946 * bytes we want to send. Many TCP/IP stacks disregard any total
947 * size parameters passed to them and just look at the frags.
949 * We always expect at least 1 mapped fragment containing the
950 * complete portals header. */
951 LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
952 lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
953 LASSERT (tx->tx_niov >= 1);
954 LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
956 CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
957 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type,
958 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
960 tx->tx_conn = NULL; /* only set when assigned a conn */
961 tx->tx_resid = tx->tx_nob;
962 tx->tx_hdr = (ptl_hdr_t *)tx->tx_iov[0].iov_base;
964 g_lock = &ksocknal_data.ksnd_global_lock;
967 peer = ksocknal_find_target_peer_locked (tx, nid);
969 read_unlock (g_lock);
970 return (-EHOSTUNREACH);
973 if (ksocknal_find_connectable_route_locked(peer) == NULL) {
974 conn = ksocknal_find_conn_locked (tx, peer);
976 /* I've got no autoconnect routes that need to be
977 * connecting and I do have an actual connection... */
978 ksocknal_queue_tx_locked (tx, conn);
979 read_unlock (g_lock);
984 /* Making one or more connections; I'll need a write lock... */
986 atomic_inc (&peer->ksnp_refcount); /* +1 ref for me while I unlock */
987 read_unlock (g_lock);
988 write_lock_irqsave (g_lock, flags);
990 if (peer->ksnp_closing) { /* peer deleted as I blocked! */
991 write_unlock_irqrestore (g_lock, flags);
992 ksocknal_put_peer (peer);
993 return (-EHOSTUNREACH);
995 ksocknal_put_peer (peer); /* drop ref I got above */
998 /* launch any/all autoconnections that need it */
999 route = ksocknal_find_connectable_route_locked (peer);
1003 ksocknal_launch_autoconnect_locked (route);
1006 conn = ksocknal_find_conn_locked (tx, peer);
1008 /* Connection exists; queue message on it */
1009 ksocknal_queue_tx_locked (tx, conn);
1010 write_unlock_irqrestore (g_lock, flags);
1014 route = ksocknal_find_connecting_route_locked (peer);
1015 if (route != NULL) {
1016 /* At least 1 connection is being established; queue the
1018 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
1019 write_unlock_irqrestore (g_lock, flags);
1023 write_unlock_irqrestore (g_lock, flags);
1024 return (-EHOSTUNREACH);
1028 ksocknal_sendmsg(nal_cb_t *nal,
1035 unsigned int payload_niov,
1036 struct iovec *payload_iov,
1037 ptl_kiov_t *payload_kiov,
1038 size_t payload_offset,
1045 /* NB 'private' is different depending on what we're sending.
1046 * Just ignore it... */
1048 CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid:"LPX64
1049 " pid %d\n", payload_nob, payload_niov, nid , pid);
1051 LASSERT (payload_nob == 0 || payload_niov > 0);
1052 LASSERT (payload_niov <= PTL_MD_MAX_IOV);
1054 /* It must be OK to kmap() if required */
1055 LASSERT (payload_kiov == NULL || !in_interrupt ());
1056 /* payload is either all vaddrs or all pages */
1057 LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
1059 if (payload_iov != NULL)
1060 desc_size = offsetof(ksock_ltx_t, ltx_iov[1 + payload_niov]);
1062 desc_size = offsetof(ksock_ltx_t, ltx_kiov[payload_niov]);
1064 if (in_interrupt() ||
1065 type == PTL_MSG_ACK ||
1066 type == PTL_MSG_REPLY) {
1067 /* Can't block if in interrupt or responding to an incoming
1069 PORTAL_ALLOC_ATOMIC(ltx, desc_size);
1071 PORTAL_ALLOC(ltx, desc_size);
1075 CERROR("Can't allocate tx desc type %d size %d %s\n",
1076 type, desc_size, in_interrupt() ? "(intr)" : "");
1077 return (PTL_NO_SPACE);
1080 atomic_inc(&ksocknal_data.ksnd_nactive_ltxs);
1082 ltx->ltx_desc_size = desc_size;
1084 /* We always have 1 mapped frag for the header */
1085 ltx->ltx_tx.tx_iov = ltx->ltx_iov;
1086 ltx->ltx_iov[0].iov_base = <x->ltx_hdr;
1087 ltx->ltx_iov[0].iov_len = sizeof(*hdr);
1088 ltx->ltx_hdr = *hdr;
1090 ltx->ltx_private = private;
1091 ltx->ltx_cookie = cookie;
1093 ltx->ltx_tx.tx_isfwd = 0;
1094 ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_nob;
1096 if (payload_iov != NULL) {
1097 /* payload is all mapped */
1098 ltx->ltx_tx.tx_kiov = NULL;
1099 ltx->ltx_tx.tx_nkiov = 0;
1101 ltx->ltx_tx.tx_niov =
1102 1 + lib_extract_iov(payload_niov, <x->ltx_iov[1],
1103 payload_niov, payload_iov,
1104 payload_offset, payload_nob);
1106 /* payload is all pages */
1107 ltx->ltx_tx.tx_niov = 1;
1109 ltx->ltx_tx.tx_kiov = ltx->ltx_kiov;
1110 ltx->ltx_tx.tx_nkiov =
1111 lib_extract_kiov(payload_niov, ltx->ltx_kiov,
1112 payload_niov, payload_kiov,
1113 payload_offset, payload_nob);
1116 rc = ksocknal_launch_packet(<x->ltx_tx, nid);
1120 ksocknal_free_ltx(ltx);
1125 ksocknal_send (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1126 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1127 unsigned int payload_niov, struct iovec *payload_iov,
1128 size_t payload_offset, size_t payload_len)
1130 return (ksocknal_sendmsg(nal, private, cookie,
1131 hdr, type, nid, pid,
1132 payload_niov, payload_iov, NULL,
1133 payload_offset, payload_len));
1137 ksocknal_send_pages (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1138 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1139 unsigned int payload_niov, ptl_kiov_t *payload_kiov,
1140 size_t payload_offset, size_t payload_len)
1142 return (ksocknal_sendmsg(nal, private, cookie,
1143 hdr, type, nid, pid,
1144 payload_niov, NULL, payload_kiov,
1145 payload_offset, payload_len));
1149 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1151 ptl_nid_t nid = fwd->kprfd_gateway_nid;
1152 ksock_ftx_t *ftx = (ksock_ftx_t *)&fwd->kprfd_scratch;
1155 CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1156 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1158 /* I'm the gateway; must be the last hop */
1159 if (nid == ksocknal_lib.ni.nid)
1160 nid = fwd->kprfd_target_nid;
1162 /* setup iov for hdr */
1163 ftx->ftx_iov.iov_base = fwd->kprfd_hdr;
1164 ftx->ftx_iov.iov_len = sizeof(ptl_hdr_t);
1166 ftx->ftx_tx.tx_isfwd = 1; /* This is a forwarding packet */
1167 ftx->ftx_tx.tx_nob = sizeof(ptl_hdr_t) + fwd->kprfd_nob;
1168 ftx->ftx_tx.tx_niov = 1;
1169 ftx->ftx_tx.tx_iov = &ftx->ftx_iov;
1170 ftx->ftx_tx.tx_nkiov = fwd->kprfd_niov;
1171 ftx->ftx_tx.tx_kiov = fwd->kprfd_kiov;
1173 rc = ksocknal_launch_packet (&ftx->ftx_tx, nid);
1175 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
1179 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1181 long pid = kernel_thread (fn, arg, 0);
1186 atomic_inc (&ksocknal_data.ksnd_nthreads);
1191 ksocknal_thread_fini (void)
1193 atomic_dec (&ksocknal_data.ksnd_nthreads);
1197 ksocknal_fmb_callback (void *arg, int error)
1199 ksock_fmb_t *fmb = (ksock_fmb_t *)arg;
1200 ksock_fmb_pool_t *fmp = fmb->fmb_pool;
1201 ptl_hdr_t *hdr = (ptl_hdr_t *)page_address(fmb->fmb_kiov[0].kiov_page);
1202 ksock_conn_t *conn = NULL;
1203 ksock_sched_t *sched;
1204 unsigned long flags;
1205 char ipbuf[PTL_NALFMT_SIZE];
1206 char ipbuf2[PTL_NALFMT_SIZE];
1209 CERROR("Failed to route packet from "
1210 LPX64" %s to "LPX64" %s: %d\n",
1211 NTOH__u64(hdr->src_nid),
1212 portals_nid2str(SOCKNAL, NTOH__u64(hdr->src_nid), ipbuf),
1213 NTOH__u64(hdr->dest_nid),
1214 portals_nid2str(SOCKNAL, NTOH__u64(hdr->dest_nid), ipbuf2),
1217 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1218 NTOH__u64 (hdr->src_nid), NTOH__u64 (hdr->dest_nid));
1220 /* drop peer ref taken on init */
1221 ksocknal_put_peer (fmb->fmb_peer);
1223 spin_lock_irqsave (&fmp->fmp_lock, flags);
1225 list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1226 fmp->fmp_nactive_fmbs--;
1228 if (!list_empty (&fmp->fmp_blocked_conns)) {
1229 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1230 ksock_conn_t, ksnc_rx_list);
1231 list_del (&conn->ksnc_rx_list);
1234 spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1239 CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1240 LASSERT (conn->ksnc_rx_scheduled);
1241 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1243 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1245 sched = conn->ksnc_scheduler;
1247 spin_lock_irqsave (&sched->kss_lock, flags);
1249 list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1250 wake_up (&sched->kss_waitq);
1252 spin_unlock_irqrestore (&sched->kss_lock, flags);
1256 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1258 int payload_nob = conn->ksnc_rx_nob_left;
1259 unsigned long flags;
1260 ksock_fmb_pool_t *pool;
1263 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1264 LASSERT (kpr_routing(&ksocknal_data.ksnd_router));
1266 if (payload_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1267 pool = &ksocknal_data.ksnd_small_fmp;
1269 pool = &ksocknal_data.ksnd_large_fmp;
1271 spin_lock_irqsave (&pool->fmp_lock, flags);
1273 if (!list_empty (&pool->fmp_idle_fmbs)) {
1274 fmb = list_entry(pool->fmp_idle_fmbs.next,
1275 ksock_fmb_t, fmb_list);
1276 list_del (&fmb->fmb_list);
1277 pool->fmp_nactive_fmbs++;
1278 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1283 /* deschedule until fmb free */
1285 conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1287 list_add_tail (&conn->ksnc_rx_list,
1288 &pool->fmp_blocked_conns);
1290 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1295 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1297 int payload_nob = conn->ksnc_rx_nob_left;
1298 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1300 int nob = payload_nob;
1302 LASSERT (conn->ksnc_rx_scheduled);
1303 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1304 LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1305 LASSERT (payload_nob >= 0);
1306 LASSERT (payload_nob <= fmb->fmb_pool->fmp_buff_pages * PAGE_SIZE);
1307 LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1308 LASSERT (fmb->fmb_kiov[0].kiov_offset == 0);
1310 /* Take a ref on the conn's peer to prevent module unload before
1311 * forwarding completes. */
1312 fmb->fmb_peer = conn->ksnc_peer;
1313 atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1315 /* Copy the header we just read into the forwarding buffer. If
1316 * there's payload, start reading reading it into the buffer,
1317 * otherwise the forwarding buffer can be kicked off
1319 fmb->fmb_hdr = conn->ksnc_hdr;
1322 LASSERT (niov < fmb->fmb_pool->fmp_buff_pages);
1323 LASSERT (fmb->fmb_kiov[niov].kiov_offset == 0);
1324 fmb->fmb_kiov[niov].kiov_len = MIN (PAGE_SIZE, nob);
1329 kpr_fwd_init(&fmb->fmb_fwd, dest_nid, &fmb->fmb_hdr,
1330 payload_nob, niov, fmb->fmb_kiov,
1331 ksocknal_fmb_callback, fmb);
1333 if (payload_nob == 0) { /* got complete packet already */
1334 CDEBUG (D_NET, "%p "LPX64"->"LPX64" fwd_start (immediate)\n",
1335 conn, NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid);
1337 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1339 ksocknal_new_packet (conn, 0); /* on to next packet */
1343 conn->ksnc_cookie = fmb; /* stash fmb for later */
1344 conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1346 /* Set up conn->ksnc_rx_kiov to read the payload into fmb's kiov-ed
1348 LASSERT (niov <= sizeof(conn->ksnc_rx_iov_space)/sizeof(ptl_kiov_t));
1350 conn->ksnc_rx_niov = 0;
1351 conn->ksnc_rx_nkiov = niov;
1352 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1353 memcpy(conn->ksnc_rx_kiov, fmb->fmb_kiov, niov * sizeof(ptl_kiov_t));
1355 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1356 NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1361 ksocknal_fwd_parse (ksock_conn_t *conn)
1364 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1365 ptl_nid_t src_nid = NTOH__u64 (conn->ksnc_hdr.src_nid);
1366 int body_len = NTOH__u32 (conn->ksnc_hdr.payload_length);
1367 char str[PTL_NALFMT_SIZE];
1368 char str2[PTL_NALFMT_SIZE];
1370 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1371 src_nid, dest_nid, conn->ksnc_rx_nob_left);
1373 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1374 LASSERT (conn->ksnc_rx_scheduled);
1376 if (body_len < 0) { /* length corrupt (overflow) */
1377 CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
1378 "packet size %d illegal\n",
1379 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1380 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1383 ksocknal_new_packet (conn, 0); /* on to new packet */
1387 if (!kpr_routing(&ksocknal_data.ksnd_router)) { /* not forwarding */
1388 CERROR("dropping packet from "LPX64" (%s) for "LPX64
1389 " (%s): not forwarding\n",
1390 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1391 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1392 /* on to new packet (skip this one's body) */
1393 ksocknal_new_packet (conn, body_len);
1397 if (body_len > PTL_MTU) { /* too big to forward */
1398 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1399 "(%s): packet size %d too big\n",
1400 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1401 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1403 /* on to new packet (skip this one's body) */
1404 ksocknal_new_packet (conn, body_len);
1408 /* should have gone direct */
1409 peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1411 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1412 "(%s): target is a peer\n",
1413 src_nid, portals_nid2str(TCPNAL, src_nid, str),
1414 dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1415 ksocknal_put_peer (peer); /* drop ref from get above */
1417 /* on to next packet (skip this one's body) */
1418 ksocknal_new_packet (conn, body_len);
1422 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB; /* Getting FMB now */
1423 conn->ksnc_rx_nob_left = body_len; /* stash packet size */
1424 conn->ksnc_rx_nob_wanted = body_len; /* (no slop) */
1428 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1430 static char ksocknal_slop_buffer[4096];
1436 if (nob_to_skip == 0) { /* right at next packet boundary now */
1437 conn->ksnc_rx_started = 0;
1438 mb (); /* racing with timeout thread */
1440 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1441 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1442 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1444 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1445 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1446 conn->ksnc_rx_iov[0].iov_len = sizeof (ptl_hdr_t);
1447 conn->ksnc_rx_niov = 1;
1449 conn->ksnc_rx_kiov = NULL;
1450 conn->ksnc_rx_nkiov = 0;
1454 /* Set up to skip as much a possible now. If there's more left
1455 * (ran out of iov entries) we'll get called again */
1457 conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1458 conn->ksnc_rx_nob_left = nob_to_skip;
1459 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1464 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1466 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1467 conn->ksnc_rx_iov[niov].iov_len = nob;
1472 } while (nob_to_skip != 0 && /* mustn't overflow conn's rx iov */
1473 niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1475 conn->ksnc_rx_niov = niov;
1476 conn->ksnc_rx_kiov = NULL;
1477 conn->ksnc_rx_nkiov = 0;
1478 conn->ksnc_rx_nob_wanted = skipped;
1483 ksocknal_process_receive (ksock_conn_t *conn)
1488 LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1490 /* doesn't need a forwarding buffer */
1491 if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1495 fmb = ksocknal_get_idle_fmb (conn);
1497 /* conn descheduled waiting for idle fmb */
1501 if (ksocknal_init_fmb (conn, fmb)) {
1502 /* packet forwarded */
1507 /* NB: sched lock NOT held */
1508 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1509 conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1510 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1511 conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1513 LASSERT (conn->ksnc_rx_nob_wanted > 0);
1515 rc = ksocknal_receive(conn);
1518 LASSERT (rc != -EAGAIN);
1521 CWARN ("[%p] EOF from "LPX64" ip %d.%d.%d.%d:%d\n",
1522 conn, conn->ksnc_peer->ksnp_nid,
1523 HIPQUAD(conn->ksnc_ipaddr),
1525 else if (!conn->ksnc_closing)
1526 CERROR ("[%p] Error %d on read from "LPX64
1527 " ip %d.%d.%d.%d:%d\n",
1528 conn, rc, conn->ksnc_peer->ksnp_nid,
1529 HIPQUAD(conn->ksnc_ipaddr),
1532 ksocknal_close_conn_and_siblings (conn, rc);
1533 return (rc == 0 ? -ESHUTDOWN : rc);
1536 if (conn->ksnc_rx_nob_wanted != 0) {
1541 switch (conn->ksnc_rx_state) {
1542 case SOCKNAL_RX_HEADER:
1543 if (conn->ksnc_hdr.type != HTON__u32(PTL_MSG_HELLO) &&
1544 NTOH__u64(conn->ksnc_hdr.dest_nid) != ksocknal_lib.ni.nid) {
1545 /* This packet isn't for me */
1546 ksocknal_fwd_parse (conn);
1547 switch (conn->ksnc_rx_state) {
1548 case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1549 return (0); /* => come back later */
1550 case SOCKNAL_RX_SLOP: /* skipping packet's body */
1551 goto try_read; /* => go read it */
1552 case SOCKNAL_RX_GET_FMB: /* forwarding */
1553 goto get_fmb; /* => go get a fwd msg buffer */
1560 /* sets wanted_len, iovs etc */
1561 lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1563 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1564 conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1565 goto try_read; /* go read the payload */
1567 /* Fall through (completed packet for me) */
1569 case SOCKNAL_RX_BODY:
1570 /* payload all received */
1571 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_OK);
1574 case SOCKNAL_RX_SLOP:
1575 /* starting new packet? */
1576 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1577 return (0); /* come back later */
1578 goto try_read; /* try to finish reading slop now */
1580 case SOCKNAL_RX_BODY_FWD:
1581 /* payload all received */
1582 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1583 conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1584 NTOH__u64 (conn->ksnc_hdr.dest_nid),
1585 conn->ksnc_rx_nob_left);
1587 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1588 * conn->ksnc_cookie */
1589 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1590 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1592 /* no slop in forwarded packets */
1593 LASSERT (conn->ksnc_rx_nob_left == 0);
1595 ksocknal_new_packet (conn, 0); /* on to next packet */
1596 return (0); /* (later) */
1604 return (-EINVAL); /* keep gcc happy */
1608 ksocknal_recv (nal_cb_t *nal, void *private, lib_msg_t *msg,
1609 unsigned int niov, struct iovec *iov,
1610 size_t offset, size_t mlen, size_t rlen)
1612 ksock_conn_t *conn = (ksock_conn_t *)private;
1614 LASSERT (mlen <= rlen);
1615 LASSERT (niov <= PTL_MD_MAX_IOV);
1617 conn->ksnc_cookie = msg;
1618 conn->ksnc_rx_nob_wanted = mlen;
1619 conn->ksnc_rx_nob_left = rlen;
1621 conn->ksnc_rx_nkiov = 0;
1622 conn->ksnc_rx_kiov = NULL;
1623 conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1624 conn->ksnc_rx_niov =
1625 lib_extract_iov(PTL_MD_MAX_IOV, conn->ksnc_rx_iov,
1626 niov, iov, offset, mlen);
1629 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1630 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1636 ksocknal_recv_pages (nal_cb_t *nal, void *private, lib_msg_t *msg,
1637 unsigned int niov, ptl_kiov_t *kiov,
1638 size_t offset, size_t mlen, size_t rlen)
1640 ksock_conn_t *conn = (ksock_conn_t *)private;
1642 LASSERT (mlen <= rlen);
1643 LASSERT (niov <= PTL_MD_MAX_IOV);
1645 conn->ksnc_cookie = msg;
1646 conn->ksnc_rx_nob_wanted = mlen;
1647 conn->ksnc_rx_nob_left = rlen;
1649 conn->ksnc_rx_niov = 0;
1650 conn->ksnc_rx_iov = NULL;
1651 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1652 conn->ksnc_rx_nkiov =
1653 lib_extract_kiov(PTL_MD_MAX_IOV, conn->ksnc_rx_kiov,
1654 niov, kiov, offset, mlen);
1657 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1658 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1663 int ksocknal_scheduler (void *arg)
1665 ksock_sched_t *sched = (ksock_sched_t *)arg;
1668 unsigned long flags;
1671 int id = sched - ksocknal_data.ksnd_schedulers;
1674 snprintf (name, sizeof (name),"ksocknald_%02d", id);
1675 kportal_daemonize (name);
1676 kportal_blockallsigs ();
1678 #if (CONFIG_SMP && CPU_AFFINITY)
1679 if ((cpu_online_map & (1 << id)) != 0) {
1681 current->cpus_allowed = (1 << id);
1683 set_cpus_allowed (current, 1<<id);
1686 CERROR ("Can't set CPU affinity for %s\n", name);
1688 #endif /* CONFIG_SMP && CPU_AFFINITY */
1690 spin_lock_irqsave (&sched->kss_lock, flags);
1692 while (!ksocknal_data.ksnd_shuttingdown) {
1693 int did_something = 0;
1695 /* Ensure I progress everything semi-fairly */
1697 if (!list_empty (&sched->kss_rx_conns)) {
1698 conn = list_entry(sched->kss_rx_conns.next,
1699 ksock_conn_t, ksnc_rx_list);
1700 list_del(&conn->ksnc_rx_list);
1702 LASSERT(conn->ksnc_rx_scheduled);
1703 LASSERT(conn->ksnc_rx_ready);
1705 /* clear rx_ready in case receive isn't complete.
1706 * Do it BEFORE we call process_recv, since
1707 * data_ready can set it any time after we release
1709 conn->ksnc_rx_ready = 0;
1710 spin_unlock_irqrestore(&sched->kss_lock, flags);
1712 rc = ksocknal_process_receive(conn);
1714 spin_lock_irqsave(&sched->kss_lock, flags);
1716 /* I'm the only one that can clear this flag */
1717 LASSERT(conn->ksnc_rx_scheduled);
1719 /* Did process_receive get everything it wanted? */
1721 conn->ksnc_rx_ready = 1;
1723 if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
1724 conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
1725 /* Conn blocked for a forwarding buffer.
1726 * It will get queued for my attention when
1727 * one becomes available (and it might just
1728 * already have been!). Meanwhile my ref
1729 * on it stays put. */
1730 } else if (conn->ksnc_rx_ready) {
1731 /* reschedule for rx */
1732 list_add_tail (&conn->ksnc_rx_list,
1733 &sched->kss_rx_conns);
1735 conn->ksnc_rx_scheduled = 0;
1737 ksocknal_put_conn(conn);
1743 if (!list_empty (&sched->kss_tx_conns)) {
1744 conn = list_entry(sched->kss_tx_conns.next,
1745 ksock_conn_t, ksnc_tx_list);
1746 list_del (&conn->ksnc_tx_list);
1748 LASSERT(conn->ksnc_tx_scheduled);
1749 LASSERT(conn->ksnc_tx_ready);
1750 LASSERT(!list_empty(&conn->ksnc_tx_queue));
1752 tx = list_entry(conn->ksnc_tx_queue.next,
1753 ksock_tx_t, tx_list);
1754 /* dequeue now so empty list => more to send */
1755 list_del(&tx->tx_list);
1757 /* Clear tx_ready in case send isn't complete. Do
1758 * it BEFORE we call process_transmit, since
1759 * write_space can set it any time after we release
1761 conn->ksnc_tx_ready = 0;
1762 spin_unlock_irqrestore (&sched->kss_lock, flags);
1764 rc = ksocknal_process_transmit(conn, tx);
1766 spin_lock_irqsave (&sched->kss_lock, flags);
1768 if (rc == -ENOMEM || rc == -EAGAIN) {
1769 /* Incomplete send: replace tx on HEAD of tx_queue */
1770 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
1772 /* Complete send; assume space for more */
1773 conn->ksnc_tx_ready = 1;
1776 if (rc == -ENOMEM) {
1777 /* Do nothing; after a short timeout, this
1778 * conn will be reposted on kss_tx_conns. */
1779 } else if (conn->ksnc_tx_ready &&
1780 !list_empty (&conn->ksnc_tx_queue)) {
1781 /* reschedule for tx */
1782 list_add_tail (&conn->ksnc_tx_list,
1783 &sched->kss_tx_conns);
1785 conn->ksnc_tx_scheduled = 0;
1787 ksocknal_put_conn (conn);
1793 if (!list_empty (&sched->kss_zctxdone_list)) {
1795 list_entry(sched->kss_zctxdone_list.next,
1796 ksock_tx_t, tx_list);
1799 list_del (&tx->tx_list);
1800 spin_unlock_irqrestore (&sched->kss_lock, flags);
1802 ksocknal_tx_done (tx, 1);
1804 spin_lock_irqsave (&sched->kss_lock, flags);
1807 if (!did_something || /* nothing to do */
1808 ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1809 spin_unlock_irqrestore (&sched->kss_lock, flags);
1813 if (!did_something) { /* wait for something to do */
1815 rc = wait_event_interruptible (sched->kss_waitq,
1816 ksocknal_data.ksnd_shuttingdown ||
1817 !list_empty(&sched->kss_rx_conns) ||
1818 !list_empty(&sched->kss_tx_conns) ||
1819 !list_empty(&sched->kss_zctxdone_list));
1821 rc = wait_event_interruptible (sched->kss_waitq,
1822 ksocknal_data.ksnd_shuttingdown ||
1823 !list_empty(&sched->kss_rx_conns) ||
1824 !list_empty(&sched->kss_tx_conns));
1830 spin_lock_irqsave (&sched->kss_lock, flags);
1834 spin_unlock_irqrestore (&sched->kss_lock, flags);
1835 ksocknal_thread_fini ();
1840 ksocknal_data_ready (struct sock *sk, int n)
1842 unsigned long flags;
1844 ksock_sched_t *sched;
1847 /* interleave correctly with closing sockets... */
1848 read_lock (&ksocknal_data.ksnd_global_lock);
1850 conn = sk->sk_user_data;
1851 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1852 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1853 sk->sk_data_ready (sk, n);
1855 sched = conn->ksnc_scheduler;
1857 spin_lock_irqsave (&sched->kss_lock, flags);
1859 conn->ksnc_rx_ready = 1;
1861 if (!conn->ksnc_rx_scheduled) { /* not being progressed */
1862 list_add_tail(&conn->ksnc_rx_list,
1863 &sched->kss_rx_conns);
1864 conn->ksnc_rx_scheduled = 1;
1865 /* extra ref for scheduler */
1866 atomic_inc (&conn->ksnc_refcount);
1868 wake_up (&sched->kss_waitq);
1871 spin_unlock_irqrestore (&sched->kss_lock, flags);
1874 read_unlock (&ksocknal_data.ksnd_global_lock);
1880 ksocknal_write_space (struct sock *sk)
1882 unsigned long flags;
1884 ksock_sched_t *sched;
1886 /* interleave correctly with closing sockets... */
1887 read_lock (&ksocknal_data.ksnd_global_lock);
1889 conn = sk->sk_user_data;
1891 CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1892 sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1893 (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1894 " ready" : " blocked"),
1895 (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1896 " scheduled" : " idle"),
1897 (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1898 " empty" : " queued"));
1900 if (conn == NULL) { /* raced with ksocknal_terminate_conn */
1901 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1902 sk->sk_write_space (sk);
1904 read_unlock (&ksocknal_data.ksnd_global_lock);
1908 if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1909 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1911 sched = conn->ksnc_scheduler;
1913 spin_lock_irqsave (&sched->kss_lock, flags);
1915 conn->ksnc_tx_ready = 1;
1917 if (!conn->ksnc_tx_scheduled && // not being progressed
1918 !list_empty(&conn->ksnc_tx_queue)){//packets to send
1919 list_add_tail (&conn->ksnc_tx_list,
1920 &sched->kss_tx_conns);
1921 conn->ksnc_tx_scheduled = 1;
1922 /* extra ref for scheduler */
1923 atomic_inc (&conn->ksnc_refcount);
1925 wake_up (&sched->kss_waitq);
1928 spin_unlock_irqrestore (&sched->kss_lock, flags);
1931 read_unlock (&ksocknal_data.ksnd_global_lock);
1935 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1938 mm_segment_t oldmm = get_fs();
1941 struct iovec iov = {
1945 struct msghdr msg = {
1950 .msg_control = NULL,
1951 .msg_controllen = 0,
1956 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1963 CERROR ("Unexpected zero rc\n");
1964 return (-ECONNABORTED);
1967 buffer = ((char *)buffer) + rc;
1975 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1978 mm_segment_t oldmm = get_fs();
1981 struct iovec iov = {
1985 struct msghdr msg = {
1990 .msg_control = NULL,
1991 .msg_controllen = 0,
1996 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
2003 return (-ECONNABORTED);
2005 buffer = ((char *)buffer) + rc;
2013 ksocknal_hello (struct socket *sock, ptl_nid_t *nid, int *type,
2018 ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
2019 char ipbuf[PTL_NALFMT_SIZE];
2020 char ipbuf2[PTL_NALFMT_SIZE];
2022 LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
2024 memset (&hdr, 0, sizeof (hdr));
2025 hmv->magic = __cpu_to_le32 (PORTALS_PROTO_MAGIC);
2026 hmv->version_major = __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR);
2027 hmv->version_minor = __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR);
2029 hdr.src_nid = __cpu_to_le64 (ksocknal_lib.ni.nid);
2030 hdr.type = __cpu_to_le32 (PTL_MSG_HELLO);
2032 hdr.msg.hello.type = __cpu_to_le32 (*type);
2033 hdr.msg.hello.incarnation =
2034 __cpu_to_le64 (ksocknal_data.ksnd_incarnation);
2036 /* Assume sufficient socket buffering for this message */
2037 rc = ksocknal_sock_write (sock, &hdr, sizeof (hdr));
2039 CERROR ("Error %d sending HELLO to "LPX64" %s\n",
2040 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2044 rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
2046 CERROR ("Error %d reading HELLO from "LPX64" %s\n",
2047 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2051 if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
2052 CERROR ("Bad magic %#08x (%#08x expected) from "LPX64" %s\n",
2053 __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, *nid,
2054 portals_nid2str(SOCKNAL, *nid, ipbuf));
2058 if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
2059 hmv->version_minor != __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
2060 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
2061 " from "LPX64" %s\n",
2062 __le16_to_cpu (hmv->version_major),
2063 __le16_to_cpu (hmv->version_minor),
2064 PORTALS_PROTO_VERSION_MAJOR,
2065 PORTALS_PROTO_VERSION_MINOR,
2066 *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2070 #if (PORTALS_PROTO_VERSION_MAJOR != 0)
2071 # error "This code only understands protocol version 0.x"
2073 /* version 0 sends magic/version as the dest_nid of a 'hello' header,
2074 * so 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 "LPX64" %s\n",
2079 rc, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf));
2083 /* ...and check we got what we expected */
2084 if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
2085 hdr.payload_length != __cpu_to_le32 (0)) {
2086 CERROR ("Expecting a HELLO hdr with 0 payload,"
2087 " but got type %d with %d payload from "LPX64" %s\n",
2088 __le32_to_cpu (hdr.type),
2089 __le32_to_cpu (hdr.payload_length), *nid,
2090 portals_nid2str(SOCKNAL, *nid, ipbuf));
2094 if (__le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
2095 CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY\n");
2099 if (*nid == PTL_NID_ANY) { /* don't know peer's nid yet */
2100 *nid = __le64_to_cpu(hdr.src_nid);
2101 } else if (*nid != __le64_to_cpu (hdr.src_nid)) {
2102 CERROR ("Connected to nid "LPX64" %s, but expecting "LPX64" %s\n",
2103 __le64_to_cpu (hdr.src_nid),
2104 portals_nid2str(SOCKNAL,
2105 __le64_to_cpu(hdr.src_nid),
2107 *nid, portals_nid2str(SOCKNAL, *nid, ipbuf2));
2111 if (*type == SOCKNAL_CONN_NONE) {
2112 /* I've accepted this connection; peer determines type */
2113 *type = __le32_to_cpu(hdr.msg.hello.type);
2115 case SOCKNAL_CONN_ANY:
2116 case SOCKNAL_CONN_CONTROL:
2118 case SOCKNAL_CONN_BULK_IN:
2119 *type = SOCKNAL_CONN_BULK_OUT;
2121 case SOCKNAL_CONN_BULK_OUT:
2122 *type = SOCKNAL_CONN_BULK_IN;
2125 CERROR ("Unexpected type %d from "LPX64" %s\n",
2127 portals_nid2str(SOCKNAL, *nid, ipbuf));
2130 } else if (__le32_to_cpu(hdr.msg.hello.type) != SOCKNAL_CONN_NONE) {
2131 CERROR ("Mismatched types: me %d "LPX64" %s %d\n",
2132 *type, *nid, portals_nid2str(SOCKNAL, *nid, ipbuf),
2133 __le32_to_cpu(hdr.msg.hello.type));
2137 *incarnation = __le64_to_cpu(hdr.msg.hello.incarnation);
2143 ksocknal_setup_sock (struct socket *sock)
2145 mm_segment_t oldmm = get_fs ();
2148 struct linger linger;
2150 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
2151 sock->sk->sk_allocation = GFP_NOFS;
2153 sock->sk->allocation = GFP_NOFS;
2156 /* Ensure this socket aborts active sends immediately when we close
2160 linger.l_linger = 0;
2163 rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2164 (char *)&linger, sizeof (linger));
2167 CERROR ("Can't set SO_LINGER: %d\n", rc);
2173 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2174 (char *)&option, sizeof (option));
2177 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2181 #if SOCKNAL_USE_KEEPALIVES
2182 /* Keepalives: If 3/4 of the timeout elapses, start probing every
2183 * second until the timeout elapses. */
2185 option = (ksocknal_tunables.ksnd_io_timeout * 3) / 4;
2187 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2188 (char *)&option, sizeof (option));
2191 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2197 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2198 (char *)&option, sizeof (option));
2201 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2205 option = ksocknal_tunables.ksnd_io_timeout / 4;
2207 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2208 (char *)&option, sizeof (option));
2211 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2217 rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
2218 (char *)&option, sizeof (option));
2221 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2229 ksocknal_connect_peer (ksock_route_t *route, int type)
2231 struct sockaddr_in peer_addr;
2232 mm_segment_t oldmm = get_fs();
2235 struct socket *sock;
2237 char ipbuf[PTL_NALFMT_SIZE];
2239 rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2241 CERROR ("Can't create autoconnect socket: %d\n", rc);
2245 /* Ugh; have to map_fd for compatibility with sockets passed in
2246 * from userspace. And we actually need the sock->file refcounting
2247 * that this gives you :) */
2249 fd = sock_map_fd (sock);
2251 sock_release (sock);
2252 CERROR ("sock_map_fd error %d\n", fd);
2256 /* NB the fd now owns the ref on sock->file */
2257 LASSERT (sock->file != NULL);
2258 LASSERT (file_count(sock->file) == 1);
2260 /* Set the socket timeouts, so our connection attempt completes in
2262 tv.tv_sec = ksocknal_tunables.ksnd_io_timeout;
2266 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2267 (char *)&tv, sizeof (tv));
2270 CERROR ("Can't set send timeout %d: %d\n",
2271 ksocknal_tunables.ksnd_io_timeout, rc);
2276 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2277 (char *)&tv, sizeof (tv));
2280 CERROR ("Can't set receive timeout %d: %d\n",
2281 ksocknal_tunables.ksnd_io_timeout, rc);
2289 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2290 (char *)&option, sizeof (option));
2293 CERROR ("Can't disable nagle: %d\n", rc);
2298 if (route->ksnr_buffer_size != 0) {
2299 int option = route->ksnr_buffer_size;
2302 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2303 (char *)&option, sizeof (option));
2306 CERROR ("Can't set send buffer %d: %d\n",
2307 route->ksnr_buffer_size, rc);
2312 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2313 (char *)&option, sizeof (option));
2316 CERROR ("Can't set receive buffer %d: %d\n",
2317 route->ksnr_buffer_size, rc);
2322 memset (&peer_addr, 0, sizeof (peer_addr));
2323 peer_addr.sin_family = AF_INET;
2324 peer_addr.sin_port = htons (route->ksnr_port);
2325 peer_addr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2327 rc = sock->ops->connect (sock, (struct sockaddr *)&peer_addr,
2328 sizeof (peer_addr), sock->file->f_flags);
2330 CERROR ("Error %d connecting to "LPX64" %s\n", rc,
2331 route->ksnr_peer->ksnp_nid,
2332 portals_nid2str(SOCKNAL,
2333 route->ksnr_peer->ksnp_nid,
2338 rc = ksocknal_create_conn (route, sock, route->ksnr_irq_affinity, type);
2340 /* Take an extra ref on sock->file to compensate for the
2341 * upcoming close which will lose fd's ref on it. */
2342 get_file (sock->file);
2351 ksocknal_autoconnect (ksock_route_t *route)
2353 LIST_HEAD (zombies);
2356 unsigned long flags;
2361 for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
2362 if ((route->ksnr_connecting & (1 << type)) != 0)
2364 LASSERT (type < SOCKNAL_CONN_NTYPES);
2366 rc = ksocknal_connect_peer (route, type);
2371 /* successfully autoconnected: create_conn did the
2372 * route/conn binding and scheduled any blocked packets */
2374 if (route->ksnr_connecting == 0) {
2375 /* No more connections required */
2380 /* Connection attempt failed */
2382 write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2384 peer = route->ksnr_peer;
2385 route->ksnr_connecting = 0;
2387 /* This is a retry rather than a new connection */
2388 LASSERT (route->ksnr_retry_interval != 0);
2389 route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2390 route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2391 SOCKNAL_MAX_RECONNECT_INTERVAL);
2393 if (!list_empty (&peer->ksnp_tx_queue) &&
2394 ksocknal_find_connecting_route_locked (peer) == NULL) {
2395 LASSERT (list_empty (&peer->ksnp_conns));
2397 /* None of the connections that the blocked packets are
2398 * waiting for have been successful. Complete them now... */
2400 tx = list_entry (peer->ksnp_tx_queue.next,
2401 ksock_tx_t, tx_list);
2402 list_del (&tx->tx_list);
2403 list_add_tail (&tx->tx_list, &zombies);
2404 } while (!list_empty (&peer->ksnp_tx_queue));
2407 /* make this route least-favourite for re-selection */
2408 if (!route->ksnr_deleted) {
2409 list_del(&route->ksnr_list);
2410 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
2413 write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2415 while (!list_empty (&zombies)) {
2416 char ipbuf[PTL_NALFMT_SIZE];
2417 char ipbuf2[PTL_NALFMT_SIZE];
2418 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2420 CERROR ("Deleting packet type %d len %d ("LPX64" %s->"LPX64" %s)\n",
2421 NTOH__u32 (tx->tx_hdr->type),
2422 NTOH__u32 (tx->tx_hdr->payload_length),
2423 NTOH__u64 (tx->tx_hdr->src_nid),
2424 portals_nid2str(SOCKNAL,
2425 NTOH__u64(tx->tx_hdr->src_nid),
2427 NTOH__u64 (tx->tx_hdr->dest_nid),
2428 portals_nid2str(SOCKNAL,
2429 NTOH__u64(tx->tx_hdr->src_nid),
2432 list_del (&tx->tx_list);
2434 ksocknal_tx_done (tx, 0);
2439 ksocknal_autoconnectd (void *arg)
2441 long id = (long)arg;
2443 unsigned long flags;
2444 ksock_route_t *route;
2447 snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2448 kportal_daemonize (name);
2449 kportal_blockallsigs ();
2451 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2453 while (!ksocknal_data.ksnd_shuttingdown) {
2455 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2456 route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2457 ksock_route_t, ksnr_connect_list);
2459 list_del (&route->ksnr_connect_list);
2460 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2462 ksocknal_autoconnect (route);
2463 ksocknal_put_route (route);
2465 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2469 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2471 rc = wait_event_interruptible (ksocknal_data.ksnd_autoconnectd_waitq,
2472 ksocknal_data.ksnd_shuttingdown ||
2473 !list_empty (&ksocknal_data.ksnd_autoconnectd_routes));
2475 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2478 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2480 ksocknal_thread_fini ();
2485 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2487 /* We're called with a shared lock on ksnd_global_lock */
2489 struct list_head *ctmp;
2490 ksock_sched_t *sched;
2492 list_for_each (ctmp, &peer->ksnp_conns) {
2493 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2494 sched = conn->ksnc_scheduler;
2496 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2497 LASSERT (!conn->ksnc_closing);
2499 if (conn->ksnc_rx_started &&
2500 time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2501 /* Timed out incomplete incoming message */
2502 atomic_inc (&conn->ksnc_refcount);
2503 CERROR ("Timed out RX from "LPX64" %p %d.%d.%d.%d\n",
2504 peer->ksnp_nid, conn, HIPQUAD(conn->ksnc_ipaddr));
2508 if ((!list_empty (&conn->ksnc_tx_queue) ||
2509 conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2510 time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2511 /* Timed out messages queued for sending, or
2512 * messages buffered in the socket's send buffer */
2513 atomic_inc (&conn->ksnc_refcount);
2514 CERROR ("Timed out TX to "LPX64" %s%d %p %d.%d.%d.%d\n",
2516 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2517 conn->ksnc_sock->sk->sk_wmem_queued, conn,
2518 HIPQUAD(conn->ksnc_ipaddr));
2527 ksocknal_check_peer_timeouts (int idx)
2529 struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2530 struct list_head *ptmp;
2535 /* NB. We expect to have a look at all the peers and not find any
2536 * connections to time out, so we just use a shared lock while we
2538 read_lock (&ksocknal_data.ksnd_global_lock);
2540 list_for_each (ptmp, peers) {
2541 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2542 conn = ksocknal_find_timed_out_conn (peer);
2545 read_unlock (&ksocknal_data.ksnd_global_lock);
2547 CERROR ("Timeout out conn->"LPX64" ip %d.%d.%d.%d:%d\n",
2549 HIPQUAD(conn->ksnc_ipaddr),
2551 ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2553 /* NB we won't find this one again, but we can't
2554 * just proceed with the next peer, since we dropped
2555 * ksnd_global_lock and it might be dead already! */
2556 ksocknal_put_conn (conn);
2561 read_unlock (&ksocknal_data.ksnd_global_lock);
2565 ksocknal_reaper (void *arg)
2568 unsigned long flags;
2570 ksock_sched_t *sched;
2571 struct list_head enomem_conns;
2576 unsigned long deadline = jiffies;
2578 kportal_daemonize ("ksocknal_reaper");
2579 kportal_blockallsigs ();
2581 INIT_LIST_HEAD(&enomem_conns);
2582 init_waitqueue_entry (&wait, current);
2584 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2586 while (!ksocknal_data.ksnd_shuttingdown) {
2588 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2589 conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2590 ksock_conn_t, ksnc_list);
2591 list_del (&conn->ksnc_list);
2593 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2595 ksocknal_terminate_conn (conn);
2596 ksocknal_put_conn (conn);
2598 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2602 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2603 conn = list_entry (ksocknal_data.ksnd_zombie_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_destroy_conn (conn);
2611 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2615 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2616 list_add(&enomem_conns, &ksocknal_data.ksnd_enomem_conns);
2617 list_del_init(&ksocknal_data.ksnd_enomem_conns);
2620 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2622 /* reschedule all the connections that stalled with ENOMEM... */
2624 while (!list_empty (&enomem_conns)) {
2625 conn = list_entry (enomem_conns.next,
2626 ksock_conn_t, ksnc_tx_list);
2627 list_del (&conn->ksnc_tx_list);
2629 sched = conn->ksnc_scheduler;
2631 spin_lock_irqsave (&sched->kss_lock, flags);
2633 LASSERT (conn->ksnc_tx_scheduled);
2634 conn->ksnc_tx_ready = 1;
2635 list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
2636 wake_up (&sched->kss_waitq);
2638 spin_unlock_irqrestore (&sched->kss_lock, flags);
2642 /* careful with the jiffy wrap... */
2643 while ((timeout = (int)(deadline - jiffies)) <= 0) {
2646 int chunk = ksocknal_data.ksnd_peer_hash_size;
2648 /* Time to check for timeouts on a few more peers: I do
2649 * checks every 'p' seconds on a proportion of the peer
2650 * table and I need to check every connection 'n' times
2651 * within a timeout interval, to ensure I detect a
2652 * timeout on any connection within (n+1)/n times the
2653 * timeout interval. */
2655 if (ksocknal_tunables.ksnd_io_timeout > n * p)
2656 chunk = (chunk * n * p) /
2657 ksocknal_tunables.ksnd_io_timeout;
2661 for (i = 0; i < chunk; i++) {
2662 ksocknal_check_peer_timeouts (peer_index);
2663 peer_index = (peer_index + 1) %
2664 ksocknal_data.ksnd_peer_hash_size;
2670 if (nenomem_conns != 0) {
2671 /* Reduce my timeout if I rescheduled ENOMEM conns.
2672 * This also prevents me getting woken immediately
2673 * if any go back on my enomem list. */
2674 timeout = SOCKNAL_ENOMEM_RETRY;
2676 ksocknal_data.ksnd_reaper_waketime = jiffies + timeout;
2678 set_current_state (TASK_INTERRUPTIBLE);
2679 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2681 if (!ksocknal_data.ksnd_shuttingdown &&
2682 list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2683 list_empty (&ksocknal_data.ksnd_zombie_conns))
2684 schedule_timeout (timeout);
2686 set_current_state (TASK_RUNNING);
2687 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2689 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2692 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2694 ksocknal_thread_fini ();
2698 nal_cb_t ksocknal_lib = {
2699 nal_data: &ksocknal_data, /* NAL private data */
2700 cb_send: ksocknal_send,
2701 cb_send_pages: ksocknal_send_pages,
2702 cb_recv: ksocknal_recv,
2703 cb_recv_pages: ksocknal_recv_pages,
2704 cb_read: ksocknal_read,
2705 cb_write: ksocknal_write,
2706 cb_malloc: ksocknal_malloc,
2707 cb_free: ksocknal_free,
2708 cb_printf: ksocknal_printf,
2709 cb_cli: ksocknal_cli,
2710 cb_sti: ksocknal_sti,
2711 cb_callback: ksocknal_callback,
2712 cb_dist: ksocknal_dist