struct iovec *iov = tx->tx_iov;
int fragsize = iov->iov_len;
unsigned long vaddr = (unsigned long)iov->iov_base;
- int more = (!list_empty (&conn->ksnc_tx_queue)) |
- (tx->tx_niov > 1) |
- (tx->tx_nkiov > 1);
+ int more = (tx->tx_niov > 1) ||
+ (tx->tx_nkiov > 0) ||
+ (!list_empty (&conn->ksnc_tx_queue));
#if SOCKNAL_ZC
int offset = vaddr & (PAGE_SIZE - 1);
int zcsize = MIN (fragsize, PAGE_SIZE - offset);
.msg_flags = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
};
mm_segment_t oldmm = get_fs();
-
+
set_fs (KERNEL_DS);
rc = sock_sendmsg(sock, &msg, fragsize);
set_fs (oldmm);
int fragsize = kiov->kiov_len;
struct page *page = kiov->kiov_page;
int offset = kiov->kiov_offset;
- int more = (!list_empty (&conn->ksnc_tx_queue)) |
- (tx->tx_nkiov > 1);
+ int more = (tx->tx_nkiov > 1) ||
+ (!list_empty (&conn->ksnc_tx_queue));
int rc;
/* NB we can't trust socket ops to either consume our iovs
* or leave them alone, so we only receive 1 frag at a time. */
LASSERT (conn->ksnc_rx_niov > 0);
LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
-
+
set_fs (KERNEL_DS);
rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
/* NB this is just a boolean............................^ */
LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
LASSERT (conn->ksnc_rx_nkiov > 0);
LASSERT (offset + fragsize <= PAGE_SIZE);
-
+
set_fs (KERNEL_DS);
rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
/* NB this is just a boolean............................^ */
if (conn->ksnc_rx_nob_wanted == 0) {
/* Completed a message segment (header or payload) */
- if (ksocknal_data.ksnd_eager_ack &&
+ if ((ksocknal_data.ksnd_eager_ack & conn->ksnc_type) != 0 &&
(conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
/* Remind the socket to ack eagerly... */
ksocknal_tx_done (tx, 0);
}
-void
-ksocknal_process_transmit (ksock_sched_t *sched, unsigned long *irq_flags)
+int
+ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
{
- ksock_conn_t *conn;
- ksock_tx_t *tx;
int rc;
-
- LASSERT (!list_empty (&sched->kss_tx_conns));
- conn = list_entry(sched->kss_tx_conns.next, ksock_conn_t, ksnc_tx_list);
- list_del (&conn->ksnc_tx_list);
-
- LASSERT (conn->ksnc_tx_scheduled);
- LASSERT (conn->ksnc_tx_ready);
- LASSERT (!list_empty (&conn->ksnc_tx_queue));
- tx = list_entry (conn->ksnc_tx_queue.next, ksock_tx_t, tx_list);
- /* assume transmit will complete now, so dequeue while I've got lock */
- list_del (&tx->tx_list);
-
- spin_unlock_irqrestore (&sched->kss_lock, *irq_flags);
-
- LASSERT (tx->tx_resid > 0);
-
- conn->ksnc_tx_ready = 0;/* write_space may race with me and set ready */
- mb(); /* => clear BEFORE trying to write */
-
+
rc = ksocknal_sendmsg (conn, tx);
CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
+ LASSERT (rc != -EAGAIN);
- if (rc != 0) {
- if (ksocknal_close_conn_unlocked (conn, rc)) {
- /* I'm the first to close */
- CERROR ("[%p] Error %d on write to "LPX64" ip %08x:%d\n",
- conn, rc, conn->ksnc_peer->ksnp_nid,
- conn->ksnc_ipaddr, conn->ksnc_port);
+ if (rc == 0) {
+ /* no errors */
+ if (tx->tx_resid != 0) {
+ /* didn't send everything */
+ return (-EAGAIN);
}
+
ksocknal_tx_launched (tx);
- spin_lock_irqsave (&sched->kss_lock, *irq_flags);
+ return (0);
+ }
- } else if (tx->tx_resid == 0) {
- /* everything went; assume more can go, and avoid
- * write_space locking */
- conn->ksnc_tx_ready = 1;
+ if (!conn->ksnc_closing)
+ CERROR ("[%p] Error %d on write to "LPX64" ip %08x:%d\n",
+ conn, rc, conn->ksnc_peer->ksnp_nid,
+ conn->ksnc_ipaddr, conn->ksnc_port);
- ksocknal_tx_launched (tx);
- spin_lock_irqsave (&sched->kss_lock, *irq_flags);
- } else {
- spin_lock_irqsave (&sched->kss_lock, *irq_flags);
+ ksocknal_close_conn_and_siblings (conn, rc);
+ ksocknal_tx_launched (tx);
- /* back onto HEAD of tx_queue */
- list_add (&tx->tx_list, &conn->ksnc_tx_queue);
- }
-
- /* no space to write, or nothing to write? */
- if (!conn->ksnc_tx_ready ||
- list_empty (&conn->ksnc_tx_queue)) {
- /* mark not scheduled */
- conn->ksnc_tx_scheduled = 0;
- /* drop scheduler's ref */
- ksocknal_put_conn (conn);
- } else {
- /* stay scheduled */
- list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
- }
-}
+ return (rc);
+}
void
ksocknal_launch_autoconnect_locked (ksock_route_t *route)
/* called holding write lock on ksnd_global_lock */
- LASSERT (route->ksnr_conn == NULL);
- LASSERT (!route->ksnr_deleted && !route->ksnr_connecting);
+ LASSERT (!route->ksnr_deleted);
+ LASSERT ((route->ksnr_connected & (1 << SOCKNAL_CONN_ANY)) == 0);
+ LASSERT ((route->ksnr_connected & KSNR_TYPED_ROUTES) != KSNR_TYPED_ROUTES);
+ LASSERT (!route->ksnr_connecting);
- route->ksnr_connecting = 1;
+ if (ksocknal_data.ksnd_typed_conns)
+ route->ksnr_connecting =
+ KSNR_TYPED_ROUTES & ~route->ksnr_connected;
+ else
+ route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
+
atomic_inc (&route->ksnr_refcount); /* extra ref for asynchd */
spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
{
struct list_head *tmp;
- ksock_conn_t *conn = NULL;
-
+ ksock_conn_t *typed = NULL;
+ int tnob = 0;
+ ksock_conn_t *fallback = NULL;
+ int fnob = 0;
+
/* Find the conn with the shortest tx queue */
list_for_each (tmp, &peer->ksnp_conns) {
- ksock_conn_t *c = list_entry (tmp, ksock_conn_t, ksnc_list);
+ ksock_conn_t *c = list_entry(tmp, ksock_conn_t, ksnc_list);
+ int nob = atomic_read(&c->ksnc_tx_nob);
LASSERT (!c->ksnc_closing);
-
- if (conn == NULL ||
- atomic_read (&conn->ksnc_tx_nob) >
- atomic_read (&c->ksnc_tx_nob))
- conn = c;
+
+ if (fallback == NULL || nob < fnob) {
+ fallback = c;
+ fnob = nob;
+ }
+
+ if (!ksocknal_data.ksnd_typed_conns)
+ continue;
+
+ switch (c->ksnc_type) {
+ default:
+ LBUG();
+ case SOCKNAL_CONN_ANY:
+ break;
+ case SOCKNAL_CONN_BULK_IN:
+ continue;
+ case SOCKNAL_CONN_BULK_OUT:
+ if (tx->tx_nob < ksocknal_data.ksnd_min_bulk)
+ continue;
+ break;
+ case SOCKNAL_CONN_CONTROL:
+ if (tx->tx_nob >= ksocknal_data.ksnd_min_bulk)
+ continue;
+ break;
+ }
+
+ if (typed == NULL || nob < tnob) {
+ typed = c;
+ tnob = nob;
+ }
}
- return (conn);
+ /* prefer the typed selection */
+ return ((typed != NULL) ? typed : fallback);
}
void
spin_lock_irqsave (&sched->kss_lock, flags);
+ conn->ksnc_tx_deadline = jiffies +
+ ksocknal_data.ksnd_io_timeout * HZ;
+ mb(); /* order with list_add_tail */
+
list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
if (conn->ksnc_tx_ready && /* able to send */
}
ksock_route_t *
-ksocknal_find_connectable_route_locked (ksock_peer_t *peer, int eager_only)
+ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
{
struct list_head *tmp;
ksock_route_t *route;
+ ksock_route_t *candidate = NULL;
+ int found = 0;
+ int bits;
list_for_each (tmp, &peer->ksnp_routes) {
route = list_entry (tmp, ksock_route_t, ksnr_list);
+ bits = route->ksnr_connected;
+
+ if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES ||
+ (bits & (1 << SOCKNAL_CONN_ANY)) != 0 ||
+ route->ksnr_connecting != 0) {
+ /* All typed connections have been established, or
+ * an untyped connection has been established, or
+ * connections are currently being established */
+ found = 1;
+ continue;
+ }
+
+ /* too soon to retry this guy? */
+ if (!time_after_eq (jiffies, route->ksnr_timeout))
+ continue;
- if (route->ksnr_conn == NULL && /* not connected */
- !route->ksnr_connecting && /* not connecting */
- (!eager_only || route->ksnr_eager) && /* wants to connect */
- time_after_eq (jiffies, route->ksnr_timeout)) /* OK to retry */
+ /* always do eager routes */
+ if (route->ksnr_eager)
return (route);
+
+ if (candidate == NULL) {
+ /* If we don't find any other route that is fully
+ * connected or connecting, the first connectable
+ * route is returned. If it fails to connect, it
+ * will get placed at the end of the list */
+ candidate = route;
+ }
}
-
- return (NULL);
+
+ return (found ? NULL : candidate);
}
ksock_route_t *
list_for_each (tmp, &peer->ksnp_routes) {
route = list_entry (tmp, ksock_route_t, ksnr_list);
- if (route->ksnr_connecting)
+ if (route->ksnr_connecting != 0)
return (route);
}
ksock_conn_t *conn;
ksock_route_t *route;
rwlock_t *g_lock;
-
+
/* Ensure the frags we've been given EXACTLY match the number of
* bytes we want to send. Many TCP/IP stacks disregard any total
* size parameters passed to them and just look at the frags.
peer = ksocknal_find_target_peer_locked (tx, nid);
if (peer == NULL) {
read_unlock (g_lock);
- return (PTL_FAIL);
+ return (-EHOSTUNREACH);
}
- if (ksocknal_find_connectable_route_locked(peer, 1) == NULL) {
+ if (ksocknal_find_connectable_route_locked(peer) == NULL) {
conn = ksocknal_find_conn_locked (tx, peer);
if (conn != NULL) {
- /* I've got no unconnected autoconnect routes that
- * need to be connected, and I do have an actual
- * connection... */
+ /* I've got no autoconnect routes that need to be
+ * connecting and I do have an actual connection... */
ksocknal_queue_tx_locked (tx, conn);
read_unlock (g_lock);
- return (PTL_OK);
+ return (0);
}
}
if (peer->ksnp_closing) { /* peer deleted as I blocked! */
write_unlock_irqrestore (g_lock, flags);
ksocknal_put_peer (peer);
- return (PTL_FAIL);
+ return (-EHOSTUNREACH);
}
ksocknal_put_peer (peer); /* drop ref I got above */
-
for (;;) {
- /* launch all eager autoconnections */
- route = ksocknal_find_connectable_route_locked (peer, 1);
+ /* launch any/all autoconnections that need it */
+ route = ksocknal_find_connectable_route_locked (peer);
if (route == NULL)
break;
/* Connection exists; queue message on it */
ksocknal_queue_tx_locked (tx, conn);
write_unlock_irqrestore (g_lock, flags);
- return (PTL_OK);
+ return (0);
}
- if (ksocknal_find_connecting_route_locked (peer) == NULL) {
- /* no autoconnect routes actually connecting now. Scrape
- * the barrel for non-eager autoconnects */
- route = ksocknal_find_connectable_route_locked (peer, 0);
- if (route != NULL) {
- ksocknal_launch_autoconnect_locked (route);
- } else {
- write_unlock_irqrestore (g_lock, flags);
- return (PTL_FAIL);
- }
+ route = ksocknal_find_connecting_route_locked (peer);
+ if (route != NULL) {
+ /* At least 1 connection is being established; queue the
+ * message... */
+ list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
+ write_unlock_irqrestore (g_lock, flags);
+ return (0);
}
-
- /* At least 1 connection is being established; queue the message... */
- list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
-
+
write_unlock_irqrestore (g_lock, flags);
- return (PTL_OK);
+ return (-EHOSTUNREACH);
}
ksock_ltx_t *
ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_len;
rc = ksocknal_launch_packet (<x->ltx_tx, nid);
- if (rc != PTL_OK)
- ksocknal_put_ltx (ltx);
+ if (rc == 0)
+ return (PTL_OK);
- return (rc);
+ ksocknal_put_ltx (ltx);
+ return (PTL_FAIL);
}
int
ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_len;
rc = ksocknal_launch_packet (<x->ltx_tx, nid);
- if (rc != PTL_OK)
- ksocknal_put_ltx (ltx);
-
- return (rc);
+ if (rc == 0)
+ return (PTL_OK);
+
+ ksocknal_put_ltx (ltx);
+ return (PTL_FAIL);
}
void
tx->tx_hdr = (ptl_hdr_t *)fwd->kprfd_iov[0].iov_base;
rc = ksocknal_launch_packet (tx, nid);
- if (rc != 0) {
- /* FIXME, could pass a better completion error */
- kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, -EHOSTUNREACH);
- }
+ if (rc != 0)
+ kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
}
int
{
ksock_peer_t *peer;
ptl_nid_t dest_nid = NTOH__u64 (conn->ksnc_hdr.dest_nid);
- int body_len = NTOH__u32 (PTL_HDR_LENGTH(&conn->ksnc_hdr));
+ ptl_nid_t src_nid = NTOH__u64 (conn->ksnc_hdr.src_nid);
+ int body_len = NTOH__u32 (conn->ksnc_hdr.payload_length);
+ char str[PTL_NALFMT_SIZE];
CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
- NTOH__u64 (conn->ksnc_hdr.src_nid),
- dest_nid, conn->ksnc_rx_nob_left);
+ src_nid, dest_nid, conn->ksnc_rx_nob_left);
LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
LASSERT (conn->ksnc_rx_scheduled);
if (body_len < 0) { /* length corrupt (overflow) */
- CERROR("dropping packet from "LPX64" for "LPX64": packet "
- "size %d illegal\n", NTOH__u64 (conn->ksnc_hdr.src_nid),
- dest_nid, body_len);
+ CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
+ "packet size %d illegal\n",
+ src_nid, portals_nid2str(TCPNAL, src_nid, str),
+ dest_nid, portals_nid2str(TCPNAL, dest_nid, str),
+ body_len);
ksocknal_new_packet (conn, 0); /* on to new packet */
- ksocknal_close_conn_unlocked (conn, -EINVAL); /* give up on conn */
return;
}
if (ksocknal_data.ksnd_fmbs == NULL) { /* not forwarding */
- CERROR("dropping packet from "LPX64" for "LPX64": not "
- "forwarding\n", conn->ksnc_hdr.src_nid,
- conn->ksnc_hdr.dest_nid);
+ CERROR("dropping packet from "LPX64" (%s) for "LPX64
+ " (%s): not forwarding\n",
+ src_nid, portals_nid2str(TCPNAL, src_nid, str),
+ dest_nid, portals_nid2str(TCPNAL, dest_nid, str));
/* on to new packet (skip this one's body) */
ksocknal_new_packet (conn, body_len);
return;
}
- if (body_len > SOCKNAL_MAX_FWD_PAYLOAD) { /* too big to forward */
- CERROR ("dropping packet from "LPX64" for "LPX64
- ": packet size %d too big\n", conn->ksnc_hdr.src_nid,
- conn->ksnc_hdr.dest_nid, body_len);
+ if (body_len > PTL_MTU) { /* too big to forward */
+ CERROR ("dropping packet from "LPX64" (%s) for "LPX64
+ "(%s): packet size %d too big\n",
+ src_nid, portals_nid2str(TCPNAL, src_nid, str),
+ dest_nid, portals_nid2str(TCPNAL, dest_nid, str),
+ body_len);
/* on to new packet (skip this one's body) */
ksocknal_new_packet (conn, body_len);
return;
/* should have gone direct */
peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
if (peer != NULL) {
- CERROR ("dropping packet from "LPX64" for "LPX64
- ": target is a peer\n", conn->ksnc_hdr.src_nid,
- conn->ksnc_hdr.dest_nid);
+ CERROR ("dropping packet from "LPX64" (%s) for "LPX64
+ "(%s): target is a peer\n",
+ src_nid, portals_nid2str(TCPNAL, src_nid, str),
+ dest_nid, portals_nid2str(TCPNAL, dest_nid, str));
ksocknal_put_peer (peer); /* drop ref from get above */
/* on to next packet (skip this one's body) */
return (0);
}
-void
-ksocknal_process_receive (ksock_sched_t *sched, unsigned long *irq_flags)
+int
+ksocknal_process_receive (ksock_conn_t *conn)
{
- ksock_conn_t *conn;
ksock_fmb_t *fmb;
int rc;
-
- /* NB: sched->ksnc_lock lock held */
-
- LASSERT (!list_empty (&sched->kss_rx_conns));
- conn = list_entry(sched->kss_rx_conns.next, ksock_conn_t, ksnc_rx_list);
- list_del (&conn->ksnc_rx_list);
-
- spin_unlock_irqrestore (&sched->kss_lock, *irq_flags);
-
- CDEBUG(D_NET, "sched %p conn %p\n", sched, conn);
+
LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
- LASSERT (conn->ksnc_rx_scheduled);
- LASSERT (conn->ksnc_rx_ready);
/* doesn't need a forwarding buffer */
if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
get_fmb:
fmb = ksocknal_get_idle_fmb (conn);
- if (fmb == NULL) { /* conn descheduled waiting for idle fmb */
- spin_lock_irqsave (&sched->kss_lock, *irq_flags);
- return;
+ if (fmb == NULL) {
+ /* conn descheduled waiting for idle fmb */
+ return (0);
}
- if (ksocknal_init_fmb (conn, fmb)) /* packet forwarded ? */
- goto out; /* come back later for next packet */
+ if (ksocknal_init_fmb (conn, fmb)) {
+ /* packet forwarded */
+ return (0);
+ }
try_read:
/* NB: sched lock NOT held */
LASSERT (conn->ksnc_rx_nob_wanted > 0);
- conn->ksnc_rx_ready = 0;/* data ready may race with me and set ready */
- mb(); /* => clear BEFORE trying to read */
-
rc = ksocknal_recvmsg(conn);
if (rc <= 0) {
- if (ksocknal_close_conn_unlocked (conn, rc)) {
- /* I'm the first to close */
- if (rc < 0)
- CERROR ("[%p] Error %d on read from "LPX64" ip %08x:%d\n",
- conn, rc, conn->ksnc_peer->ksnp_nid,
- conn->ksnc_ipaddr, conn->ksnc_port);
- else
- CWARN ("[%p] EOF from "LPX64" ip %08x:%d\n",
- conn, conn->ksnc_peer->ksnp_nid,
- conn->ksnc_ipaddr, conn->ksnc_port);
- }
- goto out;
+ if (rc == 0)
+ CWARN ("[%p] EOF from "LPX64" ip %08x:%d\n",
+ conn, conn->ksnc_peer->ksnp_nid,
+ conn->ksnc_ipaddr, conn->ksnc_port);
+ else if (!conn->ksnc_closing)
+ CERROR ("[%p] Error %d on read from "LPX64" ip %08x:%d\n",
+ conn, rc, conn->ksnc_peer->ksnp_nid,
+ conn->ksnc_ipaddr, conn->ksnc_port);
+
+ ksocknal_close_conn_and_siblings (conn, rc);
+ return (rc == 0 ? -ESHUTDOWN : rc);
}
+ if (conn->ksnc_rx_nob_wanted != 0) {
+ /* short read */
+ return (-EAGAIN);
+ }
- if (conn->ksnc_rx_nob_wanted != 0) /* short read */
- goto out; /* try again later */
-
- /* got all I wanted, assume there's more - prevent data_ready locking */
- conn->ksnc_rx_ready = 1;
-
switch (conn->ksnc_rx_state) {
case SOCKNAL_RX_HEADER:
if (conn->ksnc_hdr.type != HTON__u32(PTL_MSG_HELLO) &&
ksocknal_fwd_parse (conn);
switch (conn->ksnc_rx_state) {
case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
- goto out; /* => come back later */
+ return (0); /* => come back later */
case SOCKNAL_RX_SLOP: /* skipping packet's body */
goto try_read; /* => go read it */
case SOCKNAL_RX_GET_FMB: /* forwarding */
case SOCKNAL_RX_SLOP:
/* starting new packet? */
if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
- goto out; /* come back later */
+ return (0); /* come back later */
goto try_read; /* try to finish reading slop now */
case SOCKNAL_RX_BODY_FWD:
LASSERT (conn->ksnc_rx_nob_left == 0);
ksocknal_new_packet (conn, 0); /* on to next packet */
- goto out; /* (later) */
+ return (0); /* (later) */
default:
break;
/* Not Reached */
LBUG ();
-
- out:
- spin_lock_irqsave (&sched->kss_lock, *irq_flags);
-
- /* no data there to read? */
- if (!conn->ksnc_rx_ready) {
- /* let socket callback schedule again */
- conn->ksnc_rx_scheduled = 0;
- /* drop scheduler's ref */
- ksocknal_put_conn (conn);
- } else {
- /* stay scheduled */
- list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
- }
+ return (-EINVAL); /* keep gcc happy */
}
int
int ksocknal_scheduler (void *arg)
{
ksock_sched_t *sched = (ksock_sched_t *)arg;
+ ksock_conn_t *conn;
+ ksock_tx_t *tx;
unsigned long flags;
int rc;
int nloops = 0;
kportal_daemonize (name);
kportal_blockallsigs ();
+ current->flags |= PF_MEMALLOC;
+
#if (CONFIG_SMP && CPU_AFFINITY)
if ((cpu_online_map & (1 << id)) != 0) {
#if 1
/* Ensure I progress everything semi-fairly */
if (!list_empty (&sched->kss_rx_conns)) {
+ conn = list_entry(sched->kss_rx_conns.next,
+ ksock_conn_t, ksnc_rx_list);
+ list_del(&conn->ksnc_rx_list);
+
+ LASSERT(conn->ksnc_rx_scheduled);
+ LASSERT(conn->ksnc_rx_ready);
+
+ /* clear rx_ready in case receive isn't complete.
+ * Do it BEFORE we call process_recv, since
+ * data_ready can set it any time after we release
+ * kss_lock. */
+ conn->ksnc_rx_ready = 0;
+ spin_unlock_irqrestore(&sched->kss_lock, flags);
+
+ rc = ksocknal_process_receive(conn);
+
+ spin_lock_irqsave(&sched->kss_lock, flags);
+
+ /* I'm the only one that can clear this flag */
+ LASSERT(conn->ksnc_rx_scheduled);
+
+ /* Did process_receive get everything it wanted? */
+ if (rc == 0)
+ conn->ksnc_rx_ready = 1;
+
+ if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
+ conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
+ /* Conn blocked for a forwarding buffer.
+ * It will get queued for my attention when
+ * one becomes available (and it might just
+ * already have been!). Meanwhile my ref
+ * on it stays put. */
+ } else if (conn->ksnc_rx_ready) {
+ /* reschedule for rx */
+ list_add_tail (&conn->ksnc_rx_list,
+ &sched->kss_rx_conns);
+ } else {
+ conn->ksnc_rx_scheduled = 0;
+ /* drop my ref */
+ ksocknal_put_conn(conn);
+ }
+
did_something = 1;
- /* drops & regains kss_lock */
- ksocknal_process_receive (sched, &flags);
}
if (!list_empty (&sched->kss_tx_conns)) {
+ conn = list_entry(sched->kss_tx_conns.next,
+ ksock_conn_t, ksnc_tx_list);
+ list_del (&conn->ksnc_tx_list);
+
+ LASSERT(conn->ksnc_tx_scheduled);
+ LASSERT(conn->ksnc_tx_ready);
+ LASSERT(!list_empty(&conn->ksnc_tx_queue));
+
+ tx = list_entry(conn->ksnc_tx_queue.next,
+ ksock_tx_t, tx_list);
+ /* dequeue now so empty list => more to send */
+ list_del(&tx->tx_list);
+
+ /* Clear tx_ready in case send isn't complete. Do
+ * it BEFORE we call process_transmit, since
+ * write_space can set it any time after we release
+ * kss_lock. */
+ conn->ksnc_tx_ready = 0;
+ spin_unlock_irqrestore (&sched->kss_lock, flags);
+
+ rc = ksocknal_process_transmit(conn, tx);
+
+ spin_lock_irqsave (&sched->kss_lock, flags);
+
+ if (rc != -EAGAIN) {
+ /* error or everything went: assume more can go */
+ conn->ksnc_tx_ready = 1;
+ } else {
+ /* back onto HEAD of tx_queue */
+ list_add (&tx->tx_list, &conn->ksnc_tx_queue);
+ }
+
+ if (conn->ksnc_tx_ready &&
+ !list_empty (&conn->ksnc_tx_queue)) {
+ /* reschedule for tx */
+ list_add_tail (&conn->ksnc_tx_list,
+ &sched->kss_tx_conns);
+ } else {
+ conn->ksnc_tx_scheduled = 0;
+ /* drop my ref */
+ ksocknal_put_conn (conn);
+ }
+
did_something = 1;
- /* drops and regains kss_lock */
- ksocknal_process_transmit (sched, &flags);
}
#if SOCKNAL_ZC
if (!list_empty (&sched->kss_zctxdone_list)) {
if (conn == NULL) { /* raced with ksocknal_close_sock */
LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
sk->sk_data_ready (sk, n);
- goto out;
- }
-
- if (!conn->ksnc_rx_ready) { /* new news */
- /* Set ASAP in case of concurrent calls to me */
- conn->ksnc_rx_ready = 1;
-
+ } else {
sched = conn->ksnc_scheduler;
spin_lock_irqsave (&sched->kss_lock, flags);
- /* Set again (process_receive may have cleared while I blocked for the lock) */
conn->ksnc_rx_ready = 1;
if (!conn->ksnc_rx_scheduled) { /* not being progressed */
spin_unlock_irqrestore (&sched->kss_lock, flags);
}
- out:
read_unlock (&ksocknal_data.ksnd_global_lock);
EXIT;
if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
- if (!conn->ksnc_tx_ready) { /* new news */
- /* Set ASAP in case of concurrent calls to me */
- conn->ksnc_tx_ready = 1;
-
- sched = conn->ksnc_scheduler;
-
- spin_lock_irqsave (&sched->kss_lock, flags);
+ sched = conn->ksnc_scheduler;
- /* Set again (process_transmit may have
- cleared while I blocked for the lock) */
- conn->ksnc_tx_ready = 1;
+ spin_lock_irqsave (&sched->kss_lock, flags);
- if (!conn->ksnc_tx_scheduled && // not being progressed
- !list_empty(&conn->ksnc_tx_queue)){//packets to send
- list_add_tail (&conn->ksnc_tx_list,
- &sched->kss_tx_conns);
- conn->ksnc_tx_scheduled = 1;
- /* extra ref for scheduler */
- atomic_inc (&conn->ksnc_refcount);
+ conn->ksnc_tx_ready = 1;
- wake_up (&sched->kss_waitq);
- }
+ if (!conn->ksnc_tx_scheduled && // not being progressed
+ !list_empty(&conn->ksnc_tx_queue)){//packets to send
+ list_add_tail (&conn->ksnc_tx_list,
+ &sched->kss_tx_conns);
+ conn->ksnc_tx_scheduled = 1;
+ /* extra ref for scheduler */
+ atomic_inc (&conn->ksnc_refcount);
- spin_unlock_irqrestore (&sched->kss_lock, flags);
+ wake_up (&sched->kss_waitq);
}
+
+ spin_unlock_irqrestore (&sched->kss_lock, flags);
}
read_unlock (&ksocknal_data.ksnd_global_lock);
}
int
-ksocknal_exchange_nids (struct socket *sock, ptl_nid_t nid)
+ksocknal_hello (struct socket *sock, ptl_nid_t *nid, int *type, __u64 *incarnation)
{
int rc;
ptl_hdr_t hdr;
hdr.src_nid = __cpu_to_le64 (ksocknal_lib.ni.nid);
hdr.type = __cpu_to_le32 (PTL_MSG_HELLO);
-
+
+ hdr.msg.hello.type = __cpu_to_le32 (*type);
+ hdr.msg.hello.incarnation =
+ __cpu_to_le64 (ksocknal_data.ksnd_incarnation);
+
/* Assume sufficient socket buffering for this message */
rc = ksocknal_sock_write (sock, &hdr, sizeof (hdr));
if (rc != 0) {
- CERROR ("Error %d sending HELLO to "LPX64"\n", rc, nid);
+ CERROR ("Error %d sending HELLO to "LPX64"\n", rc, *nid);
return (rc);
}
rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
if (rc != 0) {
- CERROR ("Error %d reading HELLO from "LPX64"\n", rc, nid);
+ CERROR ("Error %d reading HELLO from "LPX64"\n", rc, *nid);
return (rc);
}
if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
CERROR ("Bad magic %#08x (%#08x expected) from "LPX64"\n",
- __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, nid);
- return (-EINVAL);
+ __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, *nid);
+ return (-EPROTO);
}
if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
__le16_to_cpu (hmv->version_minor),
PORTALS_PROTO_VERSION_MAJOR,
PORTALS_PROTO_VERSION_MINOR,
- nid);
- return (-EINVAL);
+ *nid);
+ return (-EPROTO);
}
- LASSERT (PORTALS_PROTO_VERSION_MAJOR == 0);
+#if (PORTALS_PROTO_VERSION_MAJOR != 0)
+# error "This code only understands protocol version 0.x"
+#endif
/* version 0 sends magic/version as the dest_nid of a 'hello' header,
* so read the rest of it in now... */
rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
if (rc != 0) {
CERROR ("Error %d reading rest of HELLO hdr from "LPX64"\n",
- rc, nid);
+ rc, *nid);
return (rc);
}
/* ...and check we got what we expected */
if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
- PTL_HDR_LENGTH (&hdr) != __cpu_to_le32 (0)) {
+ hdr.payload_length != __cpu_to_le32 (0)) {
CERROR ("Expecting a HELLO hdr with 0 payload,"
" but got type %d with %d payload from "LPX64"\n",
__le32_to_cpu (hdr.type),
- __le32_to_cpu (PTL_HDR_LENGTH (&hdr)), nid);
- return (-EINVAL);
+ __le32_to_cpu (hdr.payload_length), *nid);
+ return (-EPROTO);
}
-
- if (__le64_to_cpu (hdr.src_nid) != nid) {
+
+ if (__le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
+ CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY\n");
+ return (-EPROTO);
+ }
+
+ if (*nid == PTL_NID_ANY) { /* don't know peer's nid yet */
+ *nid = __le64_to_cpu(hdr.src_nid);
+ } else if (*nid != __le64_to_cpu (hdr.src_nid)) {
CERROR ("Connected to nid "LPX64", but expecting "LPX64"\n",
- __le64_to_cpu (hdr.src_nid), nid);
- return (-EINVAL);
+ __le64_to_cpu (hdr.src_nid), *nid);
+ return (-EPROTO);
+ }
+
+ if (*type == SOCKNAL_CONN_NONE) {
+ /* I've accepted this connection; peer determines type */
+ *type = __le32_to_cpu(hdr.msg.hello.type);
+ switch (*type) {
+ case SOCKNAL_CONN_ANY:
+ case SOCKNAL_CONN_CONTROL:
+ break;
+ case SOCKNAL_CONN_BULK_IN:
+ *type = SOCKNAL_CONN_BULK_OUT;
+ break;
+ case SOCKNAL_CONN_BULK_OUT:
+ *type = SOCKNAL_CONN_BULK_IN;
+ break;
+ default:
+ CERROR ("Unexpected type %d from "LPX64"\n", *type, *nid);
+ return (-EPROTO);
+ }
+ } else if (__le32_to_cpu(hdr.msg.hello.type) != SOCKNAL_CONN_NONE) {
+ CERROR ("Mismatched types: me %d "LPX64" %d\n",
+ *type, *nid, __le32_to_cpu(hdr.msg.hello.type));
+ return (-EPROTO);
}
+ *incarnation = __le64_to_cpu(hdr.msg.hello.incarnation);
+
return (0);
}
int
-ksocknal_setup_sock (struct socket *sock)
+ksocknal_setup_sock (struct socket *sock)
{
mm_segment_t oldmm = get_fs ();
int rc;
int option;
struct linger linger;
+ sock->sk->allocation = GFP_NOFS;
+
/* Ensure this socket aborts active sends immediately when we close
* it. */
-
+
linger.l_onoff = 0;
linger.l_linger = 0;
set_fs (KERNEL_DS);
- rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
+ rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
(char *)&linger, sizeof (linger));
set_fs (oldmm);
if (rc != 0) {
CERROR ("Can't set SO_LINGER: %d\n", rc);
return (rc);
}
-
+
option = -1;
set_fs (KERNEL_DS);
rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
}
int
-ksocknal_connect_peer (ksock_route_t *route)
+ksocknal_connect_peer (ksock_route_t *route, int type)
{
struct sockaddr_in peer_addr;
mm_segment_t oldmm = get_fs();
goto out;
}
- if (route->ksnr_xchange_nids) {
- rc = ksocknal_exchange_nids (sock, route->ksnr_peer->ksnp_nid);
- if (rc != 0)
- goto out;
- }
-
- rc = ksocknal_create_conn (route->ksnr_peer->ksnp_nid,
- route, sock, route->ksnr_irq_affinity);
+ rc = ksocknal_create_conn (route, sock, route->ksnr_irq_affinity, type);
if (rc == 0) {
/* Take an extra ref on sock->file to compensate for the
* upcoming close which will lose fd's ref on it. */
ksock_peer_t *peer;
unsigned long flags;
int rc;
+ int type;
- rc = ksocknal_connect_peer (route);
- if (rc == 0) {
+ for (;;) {
+ for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
+ if ((route->ksnr_connecting & (1 << type)) != 0)
+ break;
+ LASSERT (type < SOCKNAL_CONN_NTYPES);
+
+ rc = ksocknal_connect_peer (route, type);
+
+ if (rc != 0)
+ break;
+
/* successfully autoconnected: create_conn did the
- * route/conn binding and scheduled any blocked packets,
- * so there's nothing left to do now. */
- return;
+ * route/conn binding and scheduled any blocked packets */
+
+ if (route->ksnr_connecting == 0) {
+ /* No more connections required */
+ return;
+ }
}
+ /* Connection attempt failed */
+
write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
peer = route->ksnr_peer;
route->ksnr_connecting = 0;
+ /* This is a retry rather than a new connection */
LASSERT (route->ksnr_retry_interval != 0);
route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
} while (!list_empty (&peer->ksnp_tx_queue));
}
+ /* make this route least-favourite for re-selection */
+ if (!route->ksnr_deleted) {
+ list_del(&route->ksnr_list);
+ list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
+ }
+
write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
while (!list_empty (&zombies)) {
CERROR ("Deleting packet type %d len %d ("LPX64"->"LPX64")\n",
NTOH__u32 (tx->tx_hdr->type),
- NTOH__u32 (PTL_HDR_LENGTH(tx->tx_hdr)),
+ NTOH__u32 (tx->tx_hdr->payload_length),
NTOH__u64 (tx->tx_hdr->src_nid),
NTOH__u64 (tx->tx_hdr->dest_nid));
}
if ((!list_empty (&conn->ksnc_tx_queue) ||
- conn->ksnc_sock->sk->wmem_queued != 0) &&
+ conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
/* Timed out messages queued for sending, or
* messages buffered in the socket's send buffer */
CERROR ("Timed out TX to "LPX64" %s%d %p\n",
peer->ksnp_nid,
list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
- conn->ksnc_sock->sk->wmem_queued, conn);
+ conn->ksnc_sock->sk->sk_wmem_queued, conn);
return (conn);
}
}
if (conn != NULL) {
read_unlock (&ksocknal_data.ksnd_global_lock);
- if (ksocknal_close_conn_unlocked (conn, -ETIMEDOUT)) {
- /* I actually closed... */
- CERROR ("Timeout out conn->"LPX64" ip %x:%d\n",
- peer->ksnp_nid, conn->ksnc_ipaddr,
- conn->ksnc_port);
- }
-
+ CERROR ("Timeout out conn->"LPX64" ip %x:%d\n",
+ peer->ksnp_nid, conn->ksnc_ipaddr,
+ conn->ksnc_port);
+ ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
+
/* NB we won't find this one again, but we can't
* just proceed with the next peer, since we dropped
* ksnd_global_lock and it might be dead already! */
git://git.whamcloud.com / fs / lustre-release.git / blobdiff