net/smc: handle incoming CDC validation message

Call smc_cdc_msg_validate() when a CDC message with the failover
validation bit enabled was received. Validate that the sequence number
sent with the message is one we already have received. If not, messages
were lost and the connection is terminated using a new abort_work.

Signed-off-by: Karsten Graul <kgraul@linux.ibm.com>
Reviewed-by: Ursula Braun <ubraun@linux.ibm.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

diff --git a/net/smc/smc.h b/net/smc/smc.h
index 1e91137..6f1c42d 100644 (file)
--- a/net/smc/smc.h
+++ b/net/smc/smc.h
@@ -188,12 +188,14 @@ struct smc_connection {
        spinlock_t              acurs_lock;     /* protect cursors */
 #endif
        struct work_struct      close_work;     /* peer sent some closing */
+       struct work_struct      abort_work;     /* abort the connection */
        struct tasklet_struct   rx_tsklet;      /* Receiver tasklet for SMC-D */
        u8                      rx_off;         /* receive offset:
                                                 * 0 for SMC-R, 32 for SMC-D
                                                 */
        u64                     peer_token;     /* SMC-D token of peer */
        u8                      killed : 1;     /* abnormal termination */
+       u8                      out_of_sync : 1; /* out of sync with peer */
 };
 
 struct smc_sock {                              /* smc sock container */

diff --git a/net/smc/smc_cdc.c b/net/smc/smc_cdc.c
index e6b7eef..b2b85e1 100644 (file)
--- a/net/smc/smc_cdc.c
+++ b/net/smc/smc_cdc.c
@@ -282,6 +282,28 @@ static void smc_cdc_handle_urg_data_arrival(struct smc_sock *smc,
        sk_send_sigurg(&smc->sk);
 }
 
+static void smc_cdc_msg_validate(struct smc_sock *smc, struct smc_cdc_msg *cdc,
+                                struct smc_link *link)
+{
+       struct smc_connection *conn = &smc->conn;
+       u16 recv_seq = ntohs(cdc->seqno);
+       s16 diff;
+
+       /* check that seqnum was seen before */
+       diff = conn->local_rx_ctrl.seqno - recv_seq;
+       if (diff < 0) { /* diff larger than 0x7fff */
+               /* drop connection */
+               conn->out_of_sync = 1;  /* prevent any further receives */
+               spin_lock_bh(&conn->send_lock);
+               conn->local_tx_ctrl.conn_state_flags.peer_conn_abort = 1;
+               conn->lnk = link;
+               spin_unlock_bh(&conn->send_lock);
+               sock_hold(&smc->sk); /* sock_put in abort_work */
+               if (!schedule_work(&conn->abort_work))
+                       sock_put(&smc->sk);
+       }
+}
+
 static void smc_cdc_msg_recv_action(struct smc_sock *smc,
                                    struct smc_cdc_msg *cdc)
 {
@@ -412,16 +434,19 @@ static void smc_cdc_rx_handler(struct ib_wc *wc, void *buf)
        read_lock_bh(&lgr->conns_lock);
        conn = smc_lgr_find_conn(ntohl(cdc->token), lgr);
        read_unlock_bh(&lgr->conns_lock);
-       if (!conn)
+       if (!conn || conn->out_of_sync)
                return;
        smc = container_of(conn, struct smc_sock, conn);
 
-       if (!cdc->prod_flags.failover_validation) {
-               if (smc_cdc_before(ntohs(cdc->seqno),
-                                  conn->local_rx_ctrl.seqno))
-                       /* received seqno is old */
-                       return;
+       if (cdc->prod_flags.failover_validation) {
+               smc_cdc_msg_validate(smc, cdc, link);
+               return;
        }
+       if (smc_cdc_before(ntohs(cdc->seqno),
+                          conn->local_rx_ctrl.seqno))
+               /* received seqno is old */
+               return;
+
        smc_cdc_msg_recv(smc, cdc);
 }
 

diff --git a/net/smc/smc_core.c b/net/smc/smc_core.c
index a558ce0..b5633fa 100644 (file)
--- a/net/smc/smc_core.c
+++ b/net/smc/smc_core.c
@@ -615,6 +615,8 @@ void smc_conn_free(struct smc_connection *conn)
                tasklet_kill(&conn->rx_tsklet);
        } else {
                smc_cdc_tx_dismiss_slots(conn);
+               if (current_work() != &conn->abort_work)
+                       cancel_work_sync(&conn->abort_work);
        }
        if (!list_empty(&lgr->list)) {
                smc_lgr_unregister_conn(conn);
@@ -996,6 +998,18 @@ void smc_smcr_terminate_all(struct smc_ib_device *smcibdev)
        }
 }
 
+/* abort connection, abort_work scheduled from tasklet context */
+static void smc_conn_abort_work(struct work_struct *work)
+{
+       struct smc_connection *conn = container_of(work,
+                                                  struct smc_connection,
+                                                  abort_work);
+       struct smc_sock *smc = container_of(conn, struct smc_sock, conn);
+
+       smc_conn_kill(conn, true);
+       sock_put(&smc->sk); /* sock_hold done by schedulers of abort_work */
+}
+
 /* link is up - establish alternate link if applicable */
 static void smcr_link_up(struct smc_link_group *lgr,
                         struct smc_ib_device *smcibdev, u8 ibport)
@@ -1302,6 +1316,7 @@ create:
        conn->local_tx_ctrl.common.type = SMC_CDC_MSG_TYPE;
        conn->local_tx_ctrl.len = SMC_WR_TX_SIZE;
        conn->urg_state = SMC_URG_READ;
+       INIT_WORK(&smc->conn.abort_work, smc_conn_abort_work);
        if (ini->is_smcd) {
                conn->rx_off = sizeof(struct smcd_cdc_msg);
                smcd_cdc_rx_init(conn); /* init tasklet for this conn */