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_callback (nal_cb_t * nal, void *private, lib_eq_t *eq,
58 CDEBUG(D_NET, LPX64": callback eq %p ev %p\n",
61 if (eq->event_callback != NULL)
62 eq->event_callback(ev);
68 ksocknal_malloc(nal_cb_t *nal, size_t len)
72 PORTAL_ALLOC(buf, len);
81 ksocknal_free(nal_cb_t *nal, void *buf, size_t len)
83 PORTAL_FREE(buf, len);
87 ksocknal_printf(nal_cb_t *nal, const char *fmt, ...)
93 vsnprintf (msg, sizeof (msg), fmt, ap); /* sprint safely */
96 msg[sizeof (msg) - 1] = 0; /* ensure terminated */
98 CDEBUG (D_NET, "%s", msg);
102 ksocknal_cli(nal_cb_t *nal, unsigned long *flags)
104 ksock_nal_data_t *data = nal->nal_data;
106 spin_lock(&data->ksnd_nal_cb_lock);
110 ksocknal_sti(nal_cb_t *nal, unsigned long *flags)
112 ksock_nal_data_t *data;
113 data = nal->nal_data;
115 spin_unlock(&data->ksnd_nal_cb_lock);
119 ksocknal_dist(nal_cb_t *nal, ptl_nid_t nid, unsigned long *dist)
121 /* I would guess that if ksocknal_get_peer (nid) == NULL,
122 and we're not routing, then 'nid' is very distant :) */
123 if ( nal->ni.nid == nid ) {
133 ksocknal_get_ltx (int may_block)
136 ksock_ltx_t *ltx = NULL;
139 spin_lock_irqsave (&ksocknal_data.ksnd_idle_ltx_lock, flags);
141 if (!list_empty (&ksocknal_data.ksnd_idle_ltx_list)) {
142 ltx = list_entry(ksocknal_data.ksnd_idle_ltx_list.next,
143 ksock_ltx_t, ltx_tx.tx_list);
144 list_del (<x->ltx_tx.tx_list);
145 ksocknal_data.ksnd_active_ltxs++;
150 if (!list_empty(&ksocknal_data.ksnd_idle_nblk_ltx_list)) {
151 ltx = list_entry(ksocknal_data.ksnd_idle_nblk_ltx_list.next,
152 ksock_ltx_t, ltx_tx.tx_list);
153 list_del (<x->ltx_tx.tx_list);
154 ksocknal_data.ksnd_active_ltxs++;
159 spin_unlock_irqrestore(&ksocknal_data.ksnd_idle_ltx_lock,
162 wait_event (ksocknal_data.ksnd_idle_ltx_waitq,
163 !list_empty (&ksocknal_data.ksnd_idle_ltx_list));
166 spin_unlock_irqrestore (&ksocknal_data.ksnd_idle_ltx_lock, flags);
172 ksocknal_put_ltx (ksock_ltx_t *ltx)
176 spin_lock_irqsave (&ksocknal_data.ksnd_idle_ltx_lock, flags);
178 ksocknal_data.ksnd_active_ltxs--;
179 list_add_tail (<x->ltx_tx.tx_list, ltx->ltx_idle);
181 /* normal tx desc => wakeup anyone blocking for one */
182 if (ltx->ltx_idle == &ksocknal_data.ksnd_idle_ltx_list &&
183 waitqueue_active (&ksocknal_data.ksnd_idle_ltx_waitq))
184 wake_up (&ksocknal_data.ksnd_idle_ltx_waitq);
186 spin_unlock_irqrestore (&ksocknal_data.ksnd_idle_ltx_lock, flags);
191 ksocknal_kvaddr_to_page (unsigned long vaddr)
195 if (vaddr >= VMALLOC_START &&
197 page = vmalloc_to_page ((void *)vaddr);
199 else if (vaddr >= PKMAP_BASE &&
200 vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
201 page = vmalloc_to_page ((void *)vaddr);
202 /* in 2.4 ^ just walks the page tables */
205 page = virt_to_page (vaddr);
216 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
218 struct socket *sock = conn->ksnc_sock;
219 struct iovec *iov = tx->tx_iov;
220 int fragsize = iov->iov_len;
221 unsigned long vaddr = (unsigned long)iov->iov_base;
222 int more = (!list_empty (&conn->ksnc_tx_queue)) |
226 int offset = vaddr & (PAGE_SIZE - 1);
227 int zcsize = MIN (fragsize, PAGE_SIZE - offset);
232 /* NB we can't trust socket ops to either consume our iovs
233 * or leave them alone, so we only send 1 frag at a time. */
234 LASSERT (fragsize <= tx->tx_resid);
235 LASSERT (tx->tx_niov > 0);
238 if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
239 (sock->sk->route_caps & NETIF_F_SG) &&
240 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
241 (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
243 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
244 (void *)vaddr, page, page_address(page), offset, zcsize);
246 if (fragsize > zcsize) {
251 rc = tcp_sendpage_zccd(sock, page, offset, zcsize,
252 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
257 /* NB don't pass tx's iov; sendmsg may or may not update it */
258 struct iovec fragiov = { .iov_base = (void *)vaddr,
259 .iov_len = fragsize};
260 struct msghdr msg = {
267 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
269 mm_segment_t oldmm = get_fs();
272 rc = sock_sendmsg(sock, &msg, fragsize);
281 if (rc < iov->iov_len) {
282 /* didn't send whole iov entry... */
283 iov->iov_base = (void *)(vaddr + rc);
285 /* ...but did we send everything we tried to send? */
286 return ((rc == fragsize) ? 1 : -EAGAIN);
295 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
297 struct socket *sock = conn->ksnc_sock;
298 ptl_kiov_t *kiov = tx->tx_kiov;
299 int fragsize = kiov->kiov_len;
300 struct page *page = kiov->kiov_page;
301 int offset = kiov->kiov_offset;
302 int more = (!list_empty (&conn->ksnc_tx_queue)) |
306 /* NB we can't trust socket ops to either consume our iovs
307 * or leave them alone, so we only send 1 frag at a time. */
308 LASSERT (fragsize <= tx->tx_resid);
309 LASSERT (offset + fragsize <= PAGE_SIZE);
310 LASSERT (tx->tx_niov == 0);
311 LASSERT (tx->tx_nkiov > 0);
314 if (fragsize >= ksocknal_data.ksnd_zc_min_frag &&
315 (sock->sk->route_caps & NETIF_F_SG) &&
316 (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
318 CDEBUG(D_NET, "page %p + offset %x for %d\n",
319 page, offset, fragsize);
321 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
322 more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
327 char *addr = ((char *)kmap (page)) + offset;
328 struct iovec fragiov = {.iov_base = addr,
329 .iov_len = fragsize};
330 struct msghdr msg = {
337 .msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
339 mm_segment_t oldmm = get_fs();
342 rc = sock_sendmsg(sock, &msg, fragsize);
354 /* didn't send whole frag */
355 kiov->kiov_offset = offset + rc;
356 kiov->kiov_len = fragsize - rc;
360 /* everything went */
361 LASSERT (rc == fragsize);
368 ksocknal_sendmsg (ksock_conn_t *conn, ksock_tx_t *tx)
370 /* Return 0 on success, < 0 on error.
371 * caller checks tx_resid to determine progress/completion */
375 if (ksocknal_data.ksnd_stall_tx != 0) {
376 set_current_state (TASK_UNINTERRUPTIBLE);
377 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
380 rc = ksocknal_getconnsock (conn);
385 LASSERT (tx->tx_resid != 0);
387 if (conn->ksnc_closing) {
392 if (tx->tx_niov != 0)
393 rc = ksocknal_send_iov (conn, tx);
395 rc = ksocknal_send_kiov (conn, tx);
397 if (rc <= 0) { /* error or socket full? */
398 /* NB: rc == 0 and rc == -EAGAIN both mean try
399 * again later (linux stack returns -EAGAIN for
400 * this, but Adaptech TOE returns 0) */
406 /* Consider the connection alive since we managed to chuck
407 * more data into it. Really, we'd like to consider it
408 * alive only when the peer ACKs something, but
409 * write_space() only gets called back while SOCK_NOSPACE
410 * is set. Instead, we presume peer death has occurred if
411 * the socket doesn't drain within a timout */
412 conn->ksnc_tx_deadline = jiffies +
413 ksocknal_data.ksnd_io_timeout * HZ;
414 conn->ksnc_peer->ksnp_last_alive = jiffies;
416 if (tx->tx_resid == 0) { /* sent everything */
422 ksocknal_putconnsock (conn);
427 ksocknal_eager_ack (ksock_conn_t *conn)
430 mm_segment_t oldmm = get_fs();
431 struct socket *sock = conn->ksnc_sock;
433 /* Remind the socket to ACK eagerly. If I don't, the socket might
434 * think I'm about to send something it could piggy-back the ACK
435 * on, introducing delay in completing zero-copy sends in my
439 sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
440 (char *)&opt, sizeof (opt));
445 ksocknal_recv_iov (ksock_conn_t *conn)
447 struct iovec *iov = conn->ksnc_rx_iov;
448 int fragsize = iov->iov_len;
449 unsigned long vaddr = (unsigned long)iov->iov_base;
450 struct iovec fragiov = { .iov_base = (void *)vaddr,
451 .iov_len = fragsize};
452 struct msghdr msg = {
461 mm_segment_t oldmm = get_fs();
464 /* NB we can't trust socket ops to either consume our iovs
465 * or leave them alone, so we only receive 1 frag at a time. */
466 LASSERT (conn->ksnc_rx_niov > 0);
467 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
470 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
471 /* NB this is just a boolean............................^ */
477 /* received something... */
478 conn->ksnc_peer->ksnp_last_alive = jiffies;
479 conn->ksnc_rx_deadline = jiffies +
480 ksocknal_data.ksnd_io_timeout * HZ;
481 mb(); /* order with setting rx_started */
482 conn->ksnc_rx_started = 1;
484 conn->ksnc_rx_nob_wanted -= rc;
485 conn->ksnc_rx_nob_left -= rc;
488 iov->iov_base = (void *)(vaddr + rc);
489 iov->iov_len = fragsize - rc;
494 conn->ksnc_rx_niov--;
499 ksocknal_recv_kiov (ksock_conn_t *conn)
501 ptl_kiov_t *kiov = conn->ksnc_rx_kiov;
502 struct page *page = kiov->kiov_page;
503 int offset = kiov->kiov_offset;
504 int fragsize = kiov->kiov_len;
505 unsigned long vaddr = ((unsigned long)kmap (page)) + offset;
506 struct iovec fragiov = { .iov_base = (void *)vaddr,
507 .iov_len = fragsize};
508 struct msghdr msg = {
517 mm_segment_t oldmm = get_fs();
520 /* NB we can't trust socket ops to either consume our iovs
521 * or leave them alone, so we only receive 1 frag at a time. */
522 LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
523 LASSERT (conn->ksnc_rx_nkiov > 0);
524 LASSERT (offset + fragsize <= PAGE_SIZE);
527 rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
528 /* NB this is just a boolean............................^ */
536 /* received something... */
537 conn->ksnc_peer->ksnp_last_alive = jiffies;
538 conn->ksnc_rx_deadline = jiffies +
539 ksocknal_data.ksnd_io_timeout * HZ;
540 mb(); /* order with setting rx_started */
541 conn->ksnc_rx_started = 1;
543 conn->ksnc_rx_nob_wanted -= rc;
544 conn->ksnc_rx_nob_left -= rc;
547 kiov->kiov_offset = offset + rc;
548 kiov->kiov_len = fragsize - rc;
552 conn->ksnc_rx_kiov++;
553 conn->ksnc_rx_nkiov--;
558 ksocknal_recvmsg (ksock_conn_t *conn)
560 /* Return 1 on success, 0 on EOF, < 0 on error.
561 * Caller checks ksnc_rx_nob_wanted to determine
562 * progress/completion. */
566 if (ksocknal_data.ksnd_stall_rx != 0) {
567 set_current_state (TASK_UNINTERRUPTIBLE);
568 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
571 rc = ksocknal_getconnsock (conn);
576 if (conn->ksnc_closing) {
581 if (conn->ksnc_rx_niov != 0)
582 rc = ksocknal_recv_iov (conn);
584 rc = ksocknal_recv_kiov (conn);
587 /* error/EOF or partial receive */
590 } else if (rc == 0 && conn->ksnc_rx_started) {
591 /* EOF in the middle of a message */
597 /* Completed a fragment */
599 if (conn->ksnc_rx_nob_wanted == 0) {
600 /* Completed a message segment (header or payload) */
601 if (ksocknal_data.ksnd_eager_ack &&
602 (conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
603 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
604 /* Remind the socket to ack eagerly... */
605 ksocknal_eager_ack(conn);
612 ksocknal_putconnsock (conn);
618 ksocknal_zc_callback (zccd_t *zcd)
620 ksock_tx_t *tx = KSOCK_ZCCD_2_TX(zcd);
621 ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
625 /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
627 spin_lock_irqsave (&sched->kss_lock, flags);
629 list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
630 if (waitqueue_active (&sched->kss_waitq))
631 wake_up (&sched->kss_waitq);
633 spin_unlock_irqrestore (&sched->kss_lock, flags);
639 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
644 if (tx->tx_conn != NULL) {
645 /* This tx got queued on a conn; do the accounting... */
646 atomic_sub (tx->tx_nob, &tx->tx_conn->ksnc_tx_nob);
648 /* zero copy completion isn't always from
649 * process_transmit() so it needs to keep a ref on
652 ksocknal_put_conn (tx->tx_conn);
658 if (tx->tx_isfwd) { /* was a forwarded packet? */
659 kpr_fwd_done (&ksocknal_data.ksnd_router,
660 KSOCK_TX_2_KPR_FWD_DESC (tx), 0);
666 ltx = KSOCK_TX_2_KSOCK_LTX (tx);
668 lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie);
670 ksocknal_put_ltx (ltx);
675 ksocknal_tx_launched (ksock_tx_t *tx)
678 if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
679 ksock_conn_t *conn = tx->tx_conn;
681 /* zccd skbufs are still in-flight. First take a ref on
682 * conn, so it hangs about for ksocknal_tx_done... */
683 atomic_inc (&conn->ksnc_refcount);
685 /* ...then drop the initial ref on zccd, so the zero copy
686 * callback can occur */
687 zccd_put (&tx->tx_zccd);
691 /* Any zero-copy-ness (if any) has completed; I can complete the
692 * transmit now, avoiding an extra schedule */
693 ksocknal_tx_done (tx, 0);
697 ksocknal_process_transmit (ksock_sched_t *sched, unsigned long *irq_flags)
703 LASSERT (!list_empty (&sched->kss_tx_conns));
704 conn = list_entry(sched->kss_tx_conns.next, ksock_conn_t, ksnc_tx_list);
705 list_del (&conn->ksnc_tx_list);
707 LASSERT (conn->ksnc_tx_scheduled);
708 LASSERT (conn->ksnc_tx_ready);
709 LASSERT (!list_empty (&conn->ksnc_tx_queue));
710 tx = list_entry (conn->ksnc_tx_queue.next, ksock_tx_t, tx_list);
711 /* assume transmit will complete now, so dequeue while I've got lock */
712 list_del (&tx->tx_list);
714 spin_unlock_irqrestore (&sched->kss_lock, *irq_flags);
716 LASSERT (tx->tx_resid > 0);
718 conn->ksnc_tx_ready = 0;/* write_space may race with me and set ready */
719 mb(); /* => clear BEFORE trying to write */
721 rc = ksocknal_sendmsg (conn, tx);
723 CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
726 if (ksocknal_close_conn_unlocked (conn, rc)) {
727 /* I'm the first to close */
728 CERROR ("[%p] Error %d on write to "LPX64" ip %08x:%d\n",
729 conn, rc, conn->ksnc_peer->ksnp_nid,
730 conn->ksnc_ipaddr, conn->ksnc_port);
732 ksocknal_tx_launched (tx);
733 spin_lock_irqsave (&sched->kss_lock, *irq_flags);
735 } else if (tx->tx_resid == 0) {
736 /* everything went; assume more can go, and avoid
737 * write_space locking */
738 conn->ksnc_tx_ready = 1;
740 ksocknal_tx_launched (tx);
741 spin_lock_irqsave (&sched->kss_lock, *irq_flags);
743 spin_lock_irqsave (&sched->kss_lock, *irq_flags);
745 /* back onto HEAD of tx_queue */
746 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
749 /* no space to write, or nothing to write? */
750 if (!conn->ksnc_tx_ready ||
751 list_empty (&conn->ksnc_tx_queue)) {
752 /* mark not scheduled */
753 conn->ksnc_tx_scheduled = 0;
754 /* drop scheduler's ref */
755 ksocknal_put_conn (conn);
758 list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
763 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
767 /* called holding write lock on ksnd_global_lock */
769 LASSERT (route->ksnr_conn == NULL);
770 LASSERT (!route->ksnr_deleted && !route->ksnr_connecting);
772 route->ksnr_connecting = 1;
773 atomic_inc (&route->ksnr_refcount); /* extra ref for asynchd */
775 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
777 list_add_tail (&route->ksnr_connect_list,
778 &ksocknal_data.ksnd_autoconnectd_routes);
780 if (waitqueue_active (&ksocknal_data.ksnd_autoconnectd_waitq))
781 wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
783 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
787 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
789 ptl_nid_t target_nid;
791 ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
797 CERROR ("Can't send packet to "LPX64
798 ": routed target is not a peer\n", nid);
802 rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
805 CERROR ("Can't route to "LPX64": router error %d\n", nid, rc);
809 peer = ksocknal_find_peer_locked (target_nid);
813 CERROR ("Can't send packet to "LPX64": no peer entry\n", target_nid);
818 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
820 struct list_head *tmp;
821 ksock_conn_t *conn = NULL;
823 /* Find the conn with the shortest tx queue */
824 list_for_each (tmp, &peer->ksnp_conns) {
825 ksock_conn_t *c = list_entry (tmp, ksock_conn_t, ksnc_list);
827 LASSERT (!c->ksnc_closing);
830 atomic_read (&conn->ksnc_tx_nob) >
831 atomic_read (&c->ksnc_tx_nob))
839 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
842 ksock_sched_t *sched = conn->ksnc_scheduler;
844 /* called holding global lock (read or irq-write) */
846 CDEBUG (D_NET, "Sending to "LPX64" on port %d\n",
847 conn->ksnc_peer->ksnp_nid, conn->ksnc_port);
849 atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
850 tx->tx_resid = tx->tx_nob;
854 zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
855 /* NB this sets 1 ref on zccd, so the callback can only occur after
856 * I've released this ref. */
859 spin_lock_irqsave (&sched->kss_lock, flags);
861 list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
863 if (conn->ksnc_tx_ready && /* able to send */
864 !conn->ksnc_tx_scheduled) { /* not scheduled to send */
865 /* +1 ref for scheduler */
866 atomic_inc (&conn->ksnc_refcount);
867 list_add_tail (&conn->ksnc_tx_list,
868 &sched->kss_tx_conns);
869 conn->ksnc_tx_scheduled = 1;
870 if (waitqueue_active (&sched->kss_waitq))
871 wake_up (&sched->kss_waitq);
874 spin_unlock_irqrestore (&sched->kss_lock, flags);
878 ksocknal_find_connectable_route_locked (ksock_peer_t *peer, int eager_only)
880 struct list_head *tmp;
881 ksock_route_t *route;
883 list_for_each (tmp, &peer->ksnp_routes) {
884 route = list_entry (tmp, ksock_route_t, ksnr_list);
886 if (route->ksnr_conn == NULL && /* not connected */
887 !route->ksnr_connecting && /* not connecting */
888 (!eager_only || route->ksnr_eager) && /* wants to connect */
889 time_after_eq (jiffies, route->ksnr_timeout)) /* OK to retry */
897 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
899 struct list_head *tmp;
900 ksock_route_t *route;
902 list_for_each (tmp, &peer->ksnp_routes) {
903 route = list_entry (tmp, ksock_route_t, ksnr_list);
905 if (route->ksnr_connecting)
913 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
918 ksock_route_t *route;
921 /* Ensure the frags we've been given EXACTLY match the number of
922 * bytes we want to send. Many TCP/IP stacks disregard any total
923 * size parameters passed to them and just look at the frags.
925 * We always expect at least 1 mapped fragment containing the
926 * complete portals header. */
927 LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
928 lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
929 LASSERT (tx->tx_niov >= 1);
930 LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
932 CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
933 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type,
934 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
936 tx->tx_conn = NULL; /* only set when assigned a conn */
938 g_lock = &ksocknal_data.ksnd_global_lock;
941 peer = ksocknal_find_target_peer_locked (tx, nid);
943 read_unlock (g_lock);
947 if (ksocknal_find_connectable_route_locked(peer, 1) == NULL) {
948 conn = ksocknal_find_conn_locked (tx, peer);
950 /* I've got no unconnected autoconnect routes that
951 * need to be connected, and I do have an actual
953 ksocknal_queue_tx_locked (tx, conn);
954 read_unlock (g_lock);
959 /* Making one or more connections; I'll need a write lock... */
961 atomic_inc (&peer->ksnp_refcount); /* +1 ref for me while I unlock */
962 read_unlock (g_lock);
963 write_lock_irqsave (g_lock, flags);
965 if (peer->ksnp_closing) { /* peer deleted as I blocked! */
966 write_unlock_irqrestore (g_lock, flags);
967 ksocknal_put_peer (peer);
970 ksocknal_put_peer (peer); /* drop ref I got above */
974 /* launch all eager autoconnections */
975 route = ksocknal_find_connectable_route_locked (peer, 1);
979 ksocknal_launch_autoconnect_locked (route);
982 conn = ksocknal_find_conn_locked (tx, peer);
984 /* Connection exists; queue message on it */
985 ksocknal_queue_tx_locked (tx, conn);
986 write_unlock_irqrestore (g_lock, flags);
990 if (ksocknal_find_connecting_route_locked (peer) == NULL) {
991 /* no autoconnect routes actually connecting now. Scrape
992 * the barrel for non-eager autoconnects */
993 route = ksocknal_find_connectable_route_locked (peer, 0);
995 ksocknal_launch_autoconnect_locked (route);
997 write_unlock_irqrestore (g_lock, flags);
1002 /* At least 1 connection is being established; queue the message... */
1003 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
1005 write_unlock_irqrestore (g_lock, flags);
1010 ksocknal_setup_hdr (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1011 ptl_hdr_t *hdr, int type)
1015 /* I may not block for a transmit descriptor if I might block the
1016 * receiver, or an interrupt handler. */
1017 ltx = ksocknal_get_ltx (!(type == PTL_MSG_ACK ||
1018 type == PTL_MSG_REPLY ||
1021 CERROR ("Can't allocate tx desc\n");
1025 /* Init local send packet (storage for hdr, finalize() args) */
1026 ltx->ltx_hdr = *hdr;
1027 ltx->ltx_private = private;
1028 ltx->ltx_cookie = cookie;
1030 /* Init common ltx_tx */
1031 ltx->ltx_tx.tx_isfwd = 0;
1032 ltx->ltx_tx.tx_nob = sizeof (*hdr);
1034 /* We always have 1 mapped frag for the header */
1035 ltx->ltx_tx.tx_niov = 1;
1036 ltx->ltx_tx.tx_iov = <x->ltx_iov_space.hdr;
1037 ltx->ltx_tx.tx_iov[0].iov_base = <x->ltx_hdr;
1038 ltx->ltx_tx.tx_iov[0].iov_len = sizeof (ltx->ltx_hdr);
1040 ltx->ltx_tx.tx_kiov = NULL;
1041 ltx->ltx_tx.tx_nkiov = 0;
1047 ksocknal_send (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1048 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1049 unsigned int payload_niov, struct iovec *payload_iov,
1055 /* NB 'private' is different depending on what we're sending.
1056 * Just ignore it until we can rely on it
1060 "sending "LPSZ" bytes in %d mapped frags to nid: "LPX64
1061 " pid %d\n", payload_len, payload_niov, nid, pid);
1063 ltx = ksocknal_setup_hdr (nal, private, cookie, hdr, type);
1067 /* append the payload_iovs to the one pointing at the header */
1068 LASSERT (ltx->ltx_tx.tx_niov == 1 && ltx->ltx_tx.tx_nkiov == 0);
1069 LASSERT (payload_niov <= PTL_MD_MAX_IOV);
1071 memcpy (ltx->ltx_tx.tx_iov + 1, payload_iov,
1072 payload_niov * sizeof (*payload_iov));
1073 ltx->ltx_tx.tx_niov = 1 + payload_niov;
1074 ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_len;
1076 rc = ksocknal_launch_packet (<x->ltx_tx, nid);
1078 ksocknal_put_ltx (ltx);
1084 ksocknal_send_pages (nal_cb_t *nal, void *private, lib_msg_t *cookie,
1085 ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1086 unsigned int payload_niov, ptl_kiov_t *payload_iov, size_t payload_len)
1091 /* NB 'private' is different depending on what we're sending.
1092 * Just ignore it until we can rely on it */
1095 "sending "LPSZ" bytes in %d mapped frags to nid: "LPX64" pid %d\n",
1096 payload_len, payload_niov, nid, pid);
1098 ltx = ksocknal_setup_hdr (nal, private, cookie, hdr, type);
1102 LASSERT (ltx->ltx_tx.tx_niov == 1 && ltx->ltx_tx.tx_nkiov == 0);
1103 LASSERT (payload_niov <= PTL_MD_MAX_IOV);
1105 ltx->ltx_tx.tx_kiov = ltx->ltx_iov_space.payload.kiov;
1106 memcpy (ltx->ltx_tx.tx_kiov, payload_iov,
1107 payload_niov * sizeof (*payload_iov));
1108 ltx->ltx_tx.tx_nkiov = payload_niov;
1109 ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_len;
1111 rc = ksocknal_launch_packet (<x->ltx_tx, nid);
1113 ksocknal_put_ltx (ltx);
1119 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1121 ptl_nid_t nid = fwd->kprfd_gateway_nid;
1122 ksock_tx_t *tx = (ksock_tx_t *)&fwd->kprfd_scratch;
1125 CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1126 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1128 /* I'm the gateway; must be the last hop */
1129 if (nid == ksocknal_lib.ni.nid)
1130 nid = fwd->kprfd_target_nid;
1132 tx->tx_isfwd = 1; /* This is a forwarding packet */
1133 tx->tx_nob = fwd->kprfd_nob;
1134 tx->tx_niov = fwd->kprfd_niov;
1135 tx->tx_iov = fwd->kprfd_iov;
1138 tx->tx_hdr = (ptl_hdr_t *)fwd->kprfd_iov[0].iov_base;
1140 rc = ksocknal_launch_packet (tx, nid);
1142 /* FIXME, could pass a better completion error */
1143 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, -EHOSTUNREACH);
1148 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1150 long pid = kernel_thread (fn, arg, 0);
1155 atomic_inc (&ksocknal_data.ksnd_nthreads);
1160 ksocknal_thread_fini (void)
1162 atomic_dec (&ksocknal_data.ksnd_nthreads);
1166 ksocknal_fmb_callback (void *arg, int error)
1168 ksock_fmb_t *fmb = (ksock_fmb_t *)arg;
1169 ksock_fmb_pool_t *fmp = fmb->fmb_pool;
1170 ptl_hdr_t *hdr = (ptl_hdr_t *) page_address(fmb->fmb_pages[0]);
1171 ksock_conn_t *conn = NULL;
1172 ksock_sched_t *sched;
1173 unsigned long flags;
1176 CERROR("Failed to route packet from "LPX64" to "LPX64": %d\n",
1177 NTOH__u64(hdr->src_nid), NTOH__u64(hdr->dest_nid),
1180 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1181 NTOH__u64 (hdr->src_nid), NTOH__u64 (hdr->dest_nid));
1183 /* drop peer ref taken on init */
1184 ksocknal_put_peer (fmb->fmb_peer);
1186 spin_lock_irqsave (&fmp->fmp_lock, flags);
1188 list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1190 if (!list_empty (&fmp->fmp_blocked_conns)) {
1191 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1192 ksock_conn_t, ksnc_rx_list);
1193 list_del (&conn->ksnc_rx_list);
1196 spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1201 CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1202 LASSERT (conn->ksnc_rx_scheduled);
1203 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1205 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1207 sched = conn->ksnc_scheduler;
1209 spin_lock_irqsave (&sched->kss_lock, flags);
1211 list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1213 if (waitqueue_active (&sched->kss_waitq))
1214 wake_up (&sched->kss_waitq);
1216 spin_unlock_irqrestore (&sched->kss_lock, flags);
1220 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1222 int payload_nob = conn->ksnc_rx_nob_left;
1223 int packet_nob = sizeof (ptl_hdr_t) + payload_nob;
1224 unsigned long flags;
1225 ksock_fmb_pool_t *pool;
1228 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1229 LASSERT (ksocknal_data.ksnd_fmbs != NULL);
1231 if (packet_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1232 pool = &ksocknal_data.ksnd_small_fmp;
1234 pool = &ksocknal_data.ksnd_large_fmp;
1236 spin_lock_irqsave (&pool->fmp_lock, flags);
1238 if (!list_empty (&pool->fmp_idle_fmbs)) {
1239 fmb = list_entry(pool->fmp_idle_fmbs.next,
1240 ksock_fmb_t, fmb_list);
1241 list_del (&fmb->fmb_list);
1242 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1247 /* deschedule until fmb free */
1249 conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1251 list_add_tail (&conn->ksnc_rx_list,
1252 &pool->fmp_blocked_conns);
1254 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1259 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1261 int payload_nob = conn->ksnc_rx_nob_left;
1262 int packet_nob = sizeof (ptl_hdr_t) + payload_nob;
1263 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1264 int niov; /* at least the header */
1267 LASSERT (conn->ksnc_rx_scheduled);
1268 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1269 LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1270 LASSERT (payload_nob >= 0);
1271 LASSERT (packet_nob <= fmb->fmb_npages * PAGE_SIZE);
1272 LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1274 /* Got a forwarding buffer; copy the header we just read into the
1275 * forwarding buffer. If there's payload, start reading reading it
1276 * into the buffer, otherwise the forwarding buffer can be kicked
1279 * NB fmb->fmb_iov spans the WHOLE packet.
1280 * conn->ksnc_rx_iov spans just the payload.
1282 fmb->fmb_iov[0].iov_base = page_address (fmb->fmb_pages[0]);
1285 memcpy (fmb->fmb_iov[0].iov_base, &conn->ksnc_hdr, sizeof (ptl_hdr_t));
1287 /* Take a ref on the conn's peer to prevent module unload before
1288 * forwarding completes. NB we ref peer and not conn since because
1289 * all refs on conn after it has been closed must remove themselves
1291 fmb->fmb_peer = conn->ksnc_peer;
1292 atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1294 if (payload_nob == 0) { /* got complete packet already */
1295 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (immediate)\n",
1296 conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1297 dest_nid, packet_nob);
1299 fmb->fmb_iov[0].iov_len = sizeof (ptl_hdr_t);
1301 kpr_fwd_init (&fmb->fmb_fwd, dest_nid,
1302 packet_nob, 1, fmb->fmb_iov,
1303 ksocknal_fmb_callback, fmb);
1305 /* forward it now */
1306 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1308 ksocknal_new_packet (conn, 0); /* on to next packet */
1313 if (packet_nob <= PAGE_SIZE) { /* whole packet fits in first page */
1314 fmb->fmb_iov[0].iov_len = packet_nob;
1316 fmb->fmb_iov[0].iov_len = PAGE_SIZE;
1317 nob = packet_nob - PAGE_SIZE;
1320 LASSERT (niov < fmb->fmb_npages);
1321 fmb->fmb_iov[niov].iov_base =
1322 page_address (fmb->fmb_pages[niov]);
1323 fmb->fmb_iov[niov].iov_len = MIN (PAGE_SIZE, nob);
1329 kpr_fwd_init (&fmb->fmb_fwd, dest_nid,
1330 packet_nob, niov, fmb->fmb_iov,
1331 ksocknal_fmb_callback, fmb);
1333 conn->ksnc_cookie = fmb; /* stash fmb for later */
1334 conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1336 /* payload is desc's iov-ed buffer, but skipping the hdr */
1337 LASSERT (niov <= sizeof (conn->ksnc_rx_iov_space) /
1338 sizeof (struct iovec));
1340 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1341 conn->ksnc_rx_iov[0].iov_base =
1342 (void *)(((unsigned long)fmb->fmb_iov[0].iov_base) +
1343 sizeof (ptl_hdr_t));
1344 conn->ksnc_rx_iov[0].iov_len =
1345 fmb->fmb_iov[0].iov_len - sizeof (ptl_hdr_t);
1348 memcpy(&conn->ksnc_rx_iov[1], &fmb->fmb_iov[1],
1349 (niov - 1) * sizeof (struct iovec));
1351 conn->ksnc_rx_niov = niov;
1353 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1354 NTOH__u64 (conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1359 ksocknal_fwd_parse (ksock_conn_t *conn)
1362 ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
1363 int body_len = NTOH__u32 (PTL_HDR_LENGTH(&conn->ksnc_hdr));
1365 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1366 NTOH__u64 (conn->ksnc_hdr.src_nid),
1367 dest_nid, conn->ksnc_rx_nob_left);
1369 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1370 LASSERT (conn->ksnc_rx_scheduled);
1372 if (body_len < 0) { /* length corrupt (overflow) */
1373 CERROR("dropping packet from "LPX64" for "LPX64": packet "
1374 "size %d illegal\n", NTOH__u64 (conn->ksnc_hdr.src_nid),
1375 dest_nid, body_len);
1377 ksocknal_new_packet (conn, 0); /* on to new packet */
1378 ksocknal_close_conn_unlocked (conn, -EINVAL); /* give up on conn */
1382 if (ksocknal_data.ksnd_fmbs == NULL) { /* not forwarding */
1383 CERROR("dropping packet from "LPX64" for "LPX64": not "
1384 "forwarding\n", conn->ksnc_hdr.src_nid,
1385 conn->ksnc_hdr.dest_nid);
1386 /* on to new packet (skip this one's body) */
1387 ksocknal_new_packet (conn, body_len);
1391 if (body_len > SOCKNAL_MAX_FWD_PAYLOAD) { /* too big to forward */
1392 CERROR ("dropping packet from "LPX64" for "LPX64
1393 ": packet size %d too big\n", conn->ksnc_hdr.src_nid,
1394 conn->ksnc_hdr.dest_nid, body_len);
1395 /* on to new packet (skip this one's body) */
1396 ksocknal_new_packet (conn, body_len);
1400 /* should have gone direct */
1401 peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1403 CERROR ("dropping packet from "LPX64" for "LPX64
1404 ": target is a peer\n", conn->ksnc_hdr.src_nid,
1405 conn->ksnc_hdr.dest_nid);
1406 ksocknal_put_peer (peer); /* drop ref from get above */
1408 /* on to next packet (skip this one's body) */
1409 ksocknal_new_packet (conn, body_len);
1413 conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB; /* Getting FMB now */
1414 conn->ksnc_rx_nob_left = body_len; /* stash packet size */
1415 conn->ksnc_rx_nob_wanted = body_len; /* (no slop) */
1419 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1421 static char ksocknal_slop_buffer[4096];
1427 if (nob_to_skip == 0) { /* right at next packet boundary now */
1428 conn->ksnc_rx_started = 0;
1429 mb (); /* racing with timeout thread */
1431 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1432 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1433 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1435 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1436 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1437 conn->ksnc_rx_iov[0].iov_len = sizeof (ptl_hdr_t);
1438 conn->ksnc_rx_niov = 1;
1440 conn->ksnc_rx_kiov = NULL;
1441 conn->ksnc_rx_nkiov = 0;
1445 /* Set up to skip as much a possible now. If there's more left
1446 * (ran out of iov entries) we'll get called again */
1448 conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1449 conn->ksnc_rx_nob_left = nob_to_skip;
1450 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1455 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1457 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1458 conn->ksnc_rx_iov[niov].iov_len = nob;
1463 } while (nob_to_skip != 0 && /* mustn't overflow conn's rx iov */
1464 niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1466 conn->ksnc_rx_niov = niov;
1467 conn->ksnc_rx_kiov = NULL;
1468 conn->ksnc_rx_nkiov = 0;
1469 conn->ksnc_rx_nob_wanted = skipped;
1474 ksocknal_process_receive (ksock_sched_t *sched, unsigned long *irq_flags)
1480 /* NB: sched->ksnc_lock lock held */
1482 LASSERT (!list_empty (&sched->kss_rx_conns));
1483 conn = list_entry(sched->kss_rx_conns.next, ksock_conn_t, ksnc_rx_list);
1484 list_del (&conn->ksnc_rx_list);
1486 spin_unlock_irqrestore (&sched->kss_lock, *irq_flags);
1488 CDEBUG(D_NET, "sched %p conn %p\n", sched, conn);
1489 LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1490 LASSERT (conn->ksnc_rx_scheduled);
1491 LASSERT (conn->ksnc_rx_ready);
1493 /* doesn't need a forwarding buffer */
1494 if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1498 fmb = ksocknal_get_idle_fmb (conn);
1499 if (fmb == NULL) { /* conn descheduled waiting for idle fmb */
1500 spin_lock_irqsave (&sched->kss_lock, *irq_flags);
1504 if (ksocknal_init_fmb (conn, fmb)) /* packet forwarded ? */
1505 goto out; /* come back later for next packet */
1508 /* NB: sched lock NOT held */
1509 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1510 conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1511 conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1512 conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1514 LASSERT (conn->ksnc_rx_nob_wanted > 0);
1516 conn->ksnc_rx_ready = 0;/* data ready may race with me and set ready */
1517 mb(); /* => clear BEFORE trying to read */
1519 rc = ksocknal_recvmsg(conn);
1522 if (ksocknal_close_conn_unlocked (conn, rc)) {
1523 /* I'm the first to close */
1525 CERROR ("[%p] Error %d on read from "LPX64" ip %08x:%d\n",
1526 conn, rc, conn->ksnc_peer->ksnp_nid,
1527 conn->ksnc_ipaddr, conn->ksnc_port);
1529 CERROR ("[%p] EOF from "LPX64" ip %08x:%d\n",
1530 conn, conn->ksnc_peer->ksnp_nid,
1531 conn->ksnc_ipaddr, conn->ksnc_port);
1537 if (conn->ksnc_rx_nob_wanted != 0) /* short read */
1538 goto out; /* try again later */
1540 /* got all I wanted, assume there's more - prevent data_ready locking */
1541 conn->ksnc_rx_ready = 1;
1543 switch (conn->ksnc_rx_state) {
1544 case SOCKNAL_RX_HEADER:
1545 if (conn->ksnc_hdr.type != HTON__u32(PTL_MSG_HELLO) &&
1546 NTOH__u64(conn->ksnc_hdr.dest_nid) != ksocknal_lib.ni.nid) {
1547 /* This packet isn't for me */
1548 ksocknal_fwd_parse (conn);
1549 switch (conn->ksnc_rx_state) {
1550 case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1551 goto out; /* => come back later */
1552 case SOCKNAL_RX_SLOP: /* skipping packet's body */
1553 goto try_read; /* => go read it */
1554 case SOCKNAL_RX_GET_FMB: /* forwarding */
1555 goto get_fmb; /* => go get a fwd msg buffer */
1562 /* sets wanted_len, iovs etc */
1563 lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1565 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1566 conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1567 goto try_read; /* go read the payload */
1569 /* Fall through (completed packet for me) */
1571 case SOCKNAL_RX_BODY:
1572 /* payload all received */
1573 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie);
1576 case SOCKNAL_RX_SLOP:
1577 /* starting new packet? */
1578 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1579 goto out; /* come back later */
1580 goto try_read; /* try to finish reading slop now */
1582 case SOCKNAL_RX_BODY_FWD:
1583 /* payload all received */
1584 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1585 conn, NTOH__u64 (conn->ksnc_hdr.src_nid),
1586 NTOH__u64 (conn->ksnc_hdr.dest_nid),
1587 conn->ksnc_rx_nob_left);
1589 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1590 * conn->ksnc_cookie */
1591 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1592 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1594 /* no slop in forwarded packets */
1595 LASSERT (conn->ksnc_rx_nob_left == 0);
1597 ksocknal_new_packet (conn, 0); /* on to next packet */
1598 goto out; /* (later) */
1608 spin_lock_irqsave (&sched->kss_lock, *irq_flags);
1610 /* no data there to read? */
1611 if (!conn->ksnc_rx_ready) {
1612 /* let socket callback schedule again */
1613 conn->ksnc_rx_scheduled = 0;
1614 /* drop scheduler's ref */
1615 ksocknal_put_conn (conn);
1617 /* stay scheduled */
1618 list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1623 ksocknal_recv (nal_cb_t *nal, void *private, lib_msg_t *msg,
1624 unsigned int niov, struct iovec *iov, size_t mlen, size_t rlen)
1626 ksock_conn_t *conn = (ksock_conn_t *)private;
1628 LASSERT (mlen <= rlen);
1629 LASSERT (niov <= PTL_MD_MAX_IOV);
1631 conn->ksnc_cookie = msg;
1632 conn->ksnc_rx_nob_wanted = mlen;
1633 conn->ksnc_rx_nob_left = rlen;
1635 conn->ksnc_rx_nkiov = 0;
1636 conn->ksnc_rx_kiov = NULL;
1637 conn->ksnc_rx_niov = niov;
1638 conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1639 memcpy (conn->ksnc_rx_iov, iov, niov * sizeof (*iov));
1642 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1643 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1649 ksocknal_recv_pages (nal_cb_t *nal, void *private, lib_msg_t *msg,
1650 unsigned int niov, ptl_kiov_t *kiov, size_t mlen, size_t rlen)
1652 ksock_conn_t *conn = (ksock_conn_t *)private;
1654 LASSERT (mlen <= rlen);
1655 LASSERT (niov <= PTL_MD_MAX_IOV);
1657 conn->ksnc_cookie = msg;
1658 conn->ksnc_rx_nob_wanted = mlen;
1659 conn->ksnc_rx_nob_left = rlen;
1661 conn->ksnc_rx_niov = 0;
1662 conn->ksnc_rx_iov = NULL;
1663 conn->ksnc_rx_nkiov = niov;
1664 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1665 memcpy (conn->ksnc_rx_kiov, kiov, niov * sizeof (*kiov));
1668 lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1669 lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1674 int ksocknal_scheduler (void *arg)
1676 ksock_sched_t *sched = (ksock_sched_t *)arg;
1677 unsigned long flags;
1680 int id = sched - ksocknal_data.ksnd_schedulers;
1683 snprintf (name, sizeof (name),"ksocknald_%02d", id);
1684 kportal_daemonize (name);
1685 kportal_blockallsigs ();
1687 #if (CONFIG_SMP && CPU_AFFINITY)
1688 if ((cpu_online_map & (1 << id)) != 0) {
1690 current->cpus_allowed = (1 << id);
1692 set_cpus_allowed (current, 1<<id);
1695 CERROR ("Can't set CPU affinity for %s\n", name);
1697 #endif /* CONFIG_SMP && CPU_AFFINITY */
1699 spin_lock_irqsave (&sched->kss_lock, flags);
1701 while (!ksocknal_data.ksnd_shuttingdown) {
1702 int did_something = 0;
1704 /* Ensure I progress everything semi-fairly */
1706 if (!list_empty (&sched->kss_rx_conns)) {
1708 /* drops & regains kss_lock */
1709 ksocknal_process_receive (sched, &flags);
1712 if (!list_empty (&sched->kss_tx_conns)) {
1714 /* drops and regains kss_lock */
1715 ksocknal_process_transmit (sched, &flags);
1718 if (!list_empty (&sched->kss_zctxdone_list)) {
1720 list_entry(sched->kss_zctxdone_list.next,
1721 ksock_tx_t, tx_list);
1724 list_del (&tx->tx_list);
1725 spin_unlock_irqrestore (&sched->kss_lock, flags);
1727 ksocknal_tx_done (tx, 1);
1729 spin_lock_irqsave (&sched->kss_lock, flags);
1732 if (!did_something || /* nothing to do */
1733 ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1734 spin_unlock_irqrestore (&sched->kss_lock, flags);
1738 if (!did_something) { /* wait for something to do */
1740 rc = wait_event_interruptible (sched->kss_waitq,
1741 ksocknal_data.ksnd_shuttingdown ||
1742 !list_empty(&sched->kss_rx_conns) ||
1743 !list_empty(&sched->kss_tx_conns) ||
1744 !list_empty(&sched->kss_zctxdone_list));
1746 rc = wait_event_interruptible (sched->kss_waitq,
1747 ksocknal_data.ksnd_shuttingdown ||
1748 !list_empty(&sched->kss_rx_conns) ||
1749 !list_empty(&sched->kss_tx_conns));
1755 spin_lock_irqsave (&sched->kss_lock, flags);
1759 spin_unlock_irqrestore (&sched->kss_lock, flags);
1760 ksocknal_thread_fini ();
1765 ksocknal_data_ready (struct sock *sk, int n)
1767 unsigned long flags;
1769 ksock_sched_t *sched;
1772 /* interleave correctly with closing sockets... */
1773 read_lock (&ksocknal_data.ksnd_global_lock);
1775 conn = sk->sk_user_data;
1776 if (conn == NULL) { /* raced with ksocknal_close_sock */
1777 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1778 sk->sk_data_ready (sk, n);
1782 if (!conn->ksnc_rx_ready) { /* new news */
1783 /* Set ASAP in case of concurrent calls to me */
1784 conn->ksnc_rx_ready = 1;
1786 sched = conn->ksnc_scheduler;
1788 spin_lock_irqsave (&sched->kss_lock, flags);
1790 /* Set again (process_receive may have cleared while I blocked for the lock) */
1791 conn->ksnc_rx_ready = 1;
1793 if (!conn->ksnc_rx_scheduled) { /* not being progressed */
1794 list_add_tail(&conn->ksnc_rx_list,
1795 &sched->kss_rx_conns);
1796 conn->ksnc_rx_scheduled = 1;
1797 /* extra ref for scheduler */
1798 atomic_inc (&conn->ksnc_refcount);
1800 if (waitqueue_active (&sched->kss_waitq))
1801 wake_up (&sched->kss_waitq);
1804 spin_unlock_irqrestore (&sched->kss_lock, flags);
1808 read_unlock (&ksocknal_data.ksnd_global_lock);
1814 ksocknal_write_space (struct sock *sk)
1816 unsigned long flags;
1818 ksock_sched_t *sched;
1820 /* interleave correctly with closing sockets... */
1821 read_lock (&ksocknal_data.ksnd_global_lock);
1823 conn = sk->sk_user_data;
1825 CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1826 sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1827 (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1828 " ready" : " blocked"),
1829 (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1830 " scheduled" : " idle"),
1831 (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1832 " empty" : " queued"));
1834 if (conn == NULL) { /* raced with ksocknal_close_sock */
1835 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1836 sk->sk_write_space (sk);
1838 read_unlock (&ksocknal_data.ksnd_global_lock);
1842 if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1843 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1845 if (!conn->ksnc_tx_ready) { /* new news */
1846 /* Set ASAP in case of concurrent calls to me */
1847 conn->ksnc_tx_ready = 1;
1849 sched = conn->ksnc_scheduler;
1851 spin_lock_irqsave (&sched->kss_lock, flags);
1853 /* Set again (process_transmit may have
1854 cleared while I blocked for the lock) */
1855 conn->ksnc_tx_ready = 1;
1857 if (!conn->ksnc_tx_scheduled && // not being progressed
1858 !list_empty(&conn->ksnc_tx_queue)){//packets to send
1859 list_add_tail (&conn->ksnc_tx_list,
1860 &sched->kss_tx_conns);
1861 conn->ksnc_tx_scheduled = 1;
1862 /* extra ref for scheduler */
1863 atomic_inc (&conn->ksnc_refcount);
1865 if (waitqueue_active (&sched->kss_waitq))
1866 wake_up (&sched->kss_waitq);
1869 spin_unlock_irqrestore (&sched->kss_lock, flags);
1873 read_unlock (&ksocknal_data.ksnd_global_lock);
1877 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1880 mm_segment_t oldmm = get_fs();
1883 struct iovec iov = {
1887 struct msghdr msg = {
1892 .msg_control = NULL,
1893 .msg_controllen = 0,
1898 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1905 CERROR ("Unexpected zero rc\n");
1906 return (-ECONNABORTED);
1909 buffer = ((char *)buffer) + rc;
1917 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1920 mm_segment_t oldmm = get_fs();
1923 struct iovec iov = {
1927 struct msghdr msg = {
1932 .msg_control = NULL,
1933 .msg_controllen = 0,
1938 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
1945 return (-ECONNABORTED);
1947 buffer = ((char *)buffer) + rc;
1955 ksocknal_exchange_nids (struct socket *sock, ptl_nid_t nid)
1959 ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
1961 LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
1963 memset (&hdr, 0, sizeof (hdr));
1964 hmv->magic = __cpu_to_le32 (PORTALS_PROTO_MAGIC);
1965 hmv->version_major = __cpu_to_le32 (PORTALS_PROTO_VERSION_MAJOR);
1966 hmv->version_minor = __cpu_to_le32 (PORTALS_PROTO_VERSION_MINOR);
1968 hdr.src_nid = __cpu_to_le64 (ksocknal_lib.ni.nid);
1969 hdr.type = __cpu_to_le32 (PTL_MSG_HELLO);
1971 /* Assume sufficient socket buffering for this message */
1972 rc = ksocknal_sock_write (sock, &hdr, sizeof (hdr));
1974 CERROR ("Error %d sending HELLO to "LPX64"\n", rc, nid);
1978 rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
1980 CERROR ("Error %d reading HELLO from "LPX64"\n", rc, nid);
1984 if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
1985 CERROR ("Bad magic %#08x (%#08x expected) from "LPX64"\n",
1986 __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, nid);
1990 if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
1991 hmv->version_minor != __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
1992 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
1994 __le16_to_cpu (hmv->version_major),
1995 __le16_to_cpu (hmv->version_minor),
1996 PORTALS_PROTO_VERSION_MAJOR,
1997 PORTALS_PROTO_VERSION_MINOR,
2002 LASSERT (PORTALS_PROTO_VERSION_MAJOR == 0);
2003 /* version 0 sends magic/version as the dest_nid of a 'hello' header,
2004 * so read the rest of it in now... */
2006 rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
2008 CERROR ("Error %d reading rest of HELLO hdr from "LPX64"\n",
2013 /* ...and check we got what we expected */
2014 if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
2015 PTL_HDR_LENGTH (&hdr) != __cpu_to_le32 (0)) {
2016 CERROR ("Expecting a HELLO hdr with 0 payload,"
2017 " but got type %d with %d payload from "LPX64"\n",
2018 __le32_to_cpu (hdr.type),
2019 __le32_to_cpu (PTL_HDR_LENGTH (&hdr)), nid);
2023 if (__le64_to_cpu (hdr.src_nid) != nid) {
2024 CERROR ("Connected to nid "LPX64", but expecting "LPX64"\n",
2025 __le64_to_cpu (hdr.src_nid), nid);
2033 ksocknal_setup_sock (struct socket *sock)
2035 mm_segment_t oldmm = get_fs ();
2038 struct linger linger;
2040 /* Ensure this socket aborts active sends immediately when we close
2044 linger.l_linger = 0;
2047 rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2048 (char *)&linger, sizeof (linger));
2051 CERROR ("Can't set SO_LINGER: %d\n", rc);
2057 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2058 (char *)&option, sizeof (option));
2061 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2065 #if SOCKNAL_USE_KEEPALIVES
2066 /* Keepalives: If 3/4 of the timeout elapses, start probing every
2067 * second until the timeout elapses. */
2069 option = (ksocknal_data.ksnd_io_timeout * 3) / 4;
2071 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2072 (char *)&option, sizeof (option));
2075 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2081 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2082 (char *)&option, sizeof (option));
2085 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2089 option = ksocknal_data.ksnd_io_timeout / 4;
2091 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2092 (char *)&option, sizeof (option));
2095 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2101 rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
2102 (char *)&option, sizeof (option));
2105 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2113 ksocknal_connect_peer (ksock_route_t *route)
2115 struct sockaddr_in peer_addr;
2116 mm_segment_t oldmm = get_fs();
2119 struct socket *sock;
2122 rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2124 CERROR ("Can't create autoconnect socket: %d\n", rc);
2128 /* Ugh; have to map_fd for compatibility with sockets passed in
2129 * from userspace. And we actually need the sock->file refcounting
2130 * that this gives you :) */
2132 fd = sock_map_fd (sock);
2134 sock_release (sock);
2135 CERROR ("sock_map_fd error %d\n", fd);
2139 /* NB the fd now owns the ref on sock->file */
2140 LASSERT (sock->file != NULL);
2141 LASSERT (file_count(sock->file) == 1);
2143 /* Set the socket timeouts, so our connection attempt completes in
2145 tv.tv_sec = ksocknal_data.ksnd_io_timeout;
2149 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2150 (char *)&tv, sizeof (tv));
2153 CERROR ("Can't set send timeout %d: %d\n",
2154 ksocknal_data.ksnd_io_timeout, rc);
2159 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2160 (char *)&tv, sizeof (tv));
2163 CERROR ("Can't set receive timeout %d: %d\n",
2164 ksocknal_data.ksnd_io_timeout, rc);
2168 if (route->ksnr_nonagel) {
2172 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2173 (char *)&option, sizeof (option));
2176 CERROR ("Can't disable nagel: %d\n", rc);
2181 if (route->ksnr_buffer_size != 0) {
2182 int option = route->ksnr_buffer_size;
2185 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2186 (char *)&option, sizeof (option));
2189 CERROR ("Can't set send buffer %d: %d\n",
2190 route->ksnr_buffer_size, rc);
2195 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2196 (char *)&option, sizeof (option));
2199 CERROR ("Can't set receive buffer %d: %d\n",
2200 route->ksnr_buffer_size, rc);
2205 memset (&peer_addr, 0, sizeof (peer_addr));
2206 peer_addr.sin_family = AF_INET;
2207 peer_addr.sin_port = htons (route->ksnr_port);
2208 peer_addr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2210 rc = sock->ops->connect (sock, (struct sockaddr *)&peer_addr,
2211 sizeof (peer_addr), sock->file->f_flags);
2213 CERROR ("Error %d connecting to "LPX64"\n", rc,
2214 route->ksnr_peer->ksnp_nid);
2218 if (route->ksnr_xchange_nids) {
2219 rc = ksocknal_exchange_nids (sock, route->ksnr_peer->ksnp_nid);
2224 rc = ksocknal_create_conn (route->ksnr_peer->ksnp_nid,
2225 route, sock, route->ksnr_irq_affinity);
2227 /* Take an extra ref on sock->file to compensate for the
2228 * upcoming close which will lose fd's ref on it. */
2229 get_file (sock->file);
2238 ksocknal_autoconnect (ksock_route_t *route)
2240 LIST_HEAD (zombies);
2243 unsigned long flags;
2246 rc = ksocknal_connect_peer (route);
2248 /* successfully autoconnected: create_conn did the
2249 * route/conn binding and scheduled any blocked packets,
2250 * so there's nothing left to do now. */
2254 write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2256 peer = route->ksnr_peer;
2257 route->ksnr_connecting = 0;
2259 LASSERT (route->ksnr_retry_interval != 0);
2260 route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2261 route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2262 SOCKNAL_MAX_RECONNECT_INTERVAL);
2264 if (!list_empty (&peer->ksnp_tx_queue) &&
2265 ksocknal_find_connecting_route_locked (peer) == NULL) {
2266 LASSERT (list_empty (&peer->ksnp_conns));
2268 /* None of the connections that the blocked packets are
2269 * waiting for have been successful. Complete them now... */
2271 tx = list_entry (peer->ksnp_tx_queue.next,
2272 ksock_tx_t, tx_list);
2273 list_del (&tx->tx_list);
2274 list_add_tail (&tx->tx_list, &zombies);
2275 } while (!list_empty (&peer->ksnp_tx_queue));
2278 write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2280 while (!list_empty (&zombies)) {
2281 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2283 CERROR ("Deleting packet type %d len %d ("LPX64"->"LPX64")\n",
2284 NTOH__u32 (tx->tx_hdr->type),
2285 NTOH__u32 (PTL_HDR_LENGTH(tx->tx_hdr)),
2286 NTOH__u64 (tx->tx_hdr->src_nid),
2287 NTOH__u64 (tx->tx_hdr->dest_nid));
2289 list_del (&tx->tx_list);
2291 ksocknal_tx_done (tx, 0);
2296 ksocknal_autoconnectd (void *arg)
2298 long id = (long)arg;
2300 unsigned long flags;
2301 ksock_route_t *route;
2304 snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2305 kportal_daemonize (name);
2306 kportal_blockallsigs ();
2308 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2310 while (!ksocknal_data.ksnd_shuttingdown) {
2312 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2313 route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2314 ksock_route_t, ksnr_connect_list);
2316 list_del (&route->ksnr_connect_list);
2317 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2319 ksocknal_autoconnect (route);
2320 ksocknal_put_route (route);
2322 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2326 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2328 rc = wait_event_interruptible (ksocknal_data.ksnd_autoconnectd_waitq,
2329 ksocknal_data.ksnd_shuttingdown ||
2330 !list_empty (&ksocknal_data.ksnd_autoconnectd_routes));
2332 spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2335 spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2337 ksocknal_thread_fini ();
2342 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2344 /* We're called with a shared lock on ksnd_global_lock */
2346 struct list_head *ctmp;
2347 ksock_sched_t *sched;
2349 list_for_each (ctmp, &peer->ksnp_conns) {
2350 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2351 sched = conn->ksnc_scheduler;
2353 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2354 LASSERT (!conn->ksnc_closing);
2356 if (conn->ksnc_rx_started &&
2357 time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2358 /* Timed out incomplete incoming message */
2359 atomic_inc (&conn->ksnc_refcount);
2360 CERROR ("Timed out RX from "LPX64" %p\n",
2361 peer->ksnp_nid, conn);
2365 if ((!list_empty (&conn->ksnc_tx_queue) ||
2366 conn->ksnc_sock->sk->wmem_queued != 0) &&
2367 time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2368 /* Timed out messages queued for sending, or
2369 * messages buffered in the socket's send buffer */
2370 atomic_inc (&conn->ksnc_refcount);
2371 CERROR ("Timed out TX to "LPX64" %s%d %p\n",
2373 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2374 conn->ksnc_sock->sk->wmem_queued, conn);
2383 ksocknal_check_peer_timeouts (int idx)
2385 struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2386 struct list_head *ptmp;
2391 /* NB. We expect to have a look at all the peers and not find any
2392 * connections to time out, so we just use a shared lock while we
2394 read_lock (&ksocknal_data.ksnd_global_lock);
2396 list_for_each (ptmp, peers) {
2397 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2398 conn = ksocknal_find_timed_out_conn (peer);
2401 read_unlock (&ksocknal_data.ksnd_global_lock);
2403 if (ksocknal_close_conn_unlocked (conn, -ETIMEDOUT)) {
2404 /* I actually closed... */
2405 CERROR ("Timeout out conn->"LPX64" ip %x:%d\n",
2406 peer->ksnp_nid, conn->ksnc_ipaddr,
2410 /* NB we won't find this one again, but we can't
2411 * just proceed with the next peer, since we dropped
2412 * ksnd_global_lock and it might be dead already! */
2413 ksocknal_put_conn (conn);
2418 read_unlock (&ksocknal_data.ksnd_global_lock);
2422 ksocknal_reaper (void *arg)
2425 unsigned long flags;
2430 unsigned long deadline = jiffies;
2432 kportal_daemonize ("ksocknal_reaper");
2433 kportal_blockallsigs ();
2435 init_waitqueue_entry (&wait, current);
2437 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2439 while (!ksocknal_data.ksnd_shuttingdown) {
2441 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2442 conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2443 ksock_conn_t, ksnc_list);
2444 list_del (&conn->ksnc_list);
2446 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2448 ksocknal_terminate_conn (conn);
2449 ksocknal_put_conn (conn);
2451 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2455 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2456 conn = list_entry (ksocknal_data.ksnd_zombie_conns.next,
2457 ksock_conn_t, ksnc_list);
2458 list_del (&conn->ksnc_list);
2460 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2462 ksocknal_destroy_conn (conn);
2464 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2468 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2470 /* careful with the jiffy wrap... */
2471 while ((timeout = ((int)deadline - (int)jiffies)) <= 0) {
2474 int chunk = ksocknal_data.ksnd_peer_hash_size;
2476 /* Time to check for timeouts on a few more peers: I do
2477 * checks every 'p' seconds on a proportion of the peer
2478 * table and I need to check every connection 'n' times
2479 * within a timeout interval, to ensure I detect a
2480 * timeout on any connection within (n+1)/n times the
2481 * timeout interval. */
2483 if (ksocknal_data.ksnd_io_timeout > n * p)
2484 chunk = (chunk * n * p) /
2485 ksocknal_data.ksnd_io_timeout;
2489 for (i = 0; i < chunk; i++) {
2490 ksocknal_check_peer_timeouts (peer_index);
2491 peer_index = (peer_index + 1) %
2492 ksocknal_data.ksnd_peer_hash_size;
2498 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2499 set_current_state (TASK_INTERRUPTIBLE);
2501 if (!ksocknal_data.ksnd_shuttingdown &&
2502 list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2503 list_empty (&ksocknal_data.ksnd_zombie_conns))
2504 schedule_timeout (timeout);
2506 set_current_state (TASK_RUNNING);
2507 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2509 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2512 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2514 ksocknal_thread_fini ();
2518 nal_cb_t ksocknal_lib = {
2519 nal_data: &ksocknal_data, /* NAL private data */
2520 cb_send: ksocknal_send,
2521 cb_send_pages: ksocknal_send_pages,
2522 cb_recv: ksocknal_recv,
2523 cb_recv_pages: ksocknal_recv_pages,
2524 cb_read: ksocknal_read,
2525 cb_write: ksocknal_write,
2526 cb_callback: ksocknal_callback,
2527 cb_malloc: ksocknal_malloc,
2528 cb_free: ksocknal_free,
2529 cb_printf: ksocknal_printf,
2530 cb_cli: ksocknal_cli,
2531 cb_sti: ksocknal_sti,
2532 cb_dist: ksocknal_dist