another batch of messenger patches

Index: b/net/ceph/messenger.c
===================================================================
@@ -1472,7 +1472,7 @@ static int process_banner(struct ceph_co
 		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
 	}
 
-	return 0;
+	return 1;
 }
 
 static void fail_protocol(struct ceph_connection *con)

@@ -1970,6 +1970,57 @@ static void process_message(struct ceph_
 	prepare_read_tag(con);
 }
 
+/*
+ * Initiate the first phase of establishing a connection with
+ * the peer (connecting).  This phase consists of:
+ *     - client requests TCP connection to server
+ *     - server accepts TCP connection from client
+ *     - client sends banner to server
+ *     - server receives and validates client's banner
+ *     - client sends little-endian encoded own socket (IP) address
+ *     - server recieves, validates, and records client's encoded address
+ * If all is well to this point, then we begin processing the
+ * connect response.
+ */
+static int ceph_con_connect(struct ceph_connection *con)
+{
+	set_bit(CONNECTING, &con->state);
+
+	con_out_kvec_reset(con);
+	prepare_write_banner(con);
+	prepare_read_banner(con);
+
+	BUG_ON(con->in_msg);
+	con->in_tag = CEPH_MSGR_TAG_READY;
+	dout("%s initiating connect on %p new state %lu\n",
+		__func__, con, con->state);
+
+	return ceph_tcp_connect(con);
+}
+
+/*
+ * Handle the response from the first phase of establishing a
+ * connection with the peer.  This consists of:
+ *     - server sends banner to client
+ *     - client receives and validates server's banner
+ *     - server sends little-endian encoded own socket (IP) address
+ *     - client recieves, validates, and records server's encoded address
+ *     - server sends little-endian encoded socket (IP) address for client
+ *     - client recieves and records its encoded address supplied by server
+ * If all is well to this point, then we can transition to the
+ * NEGOTIATING state.
+ */
+static int ceph_con_connect_response(struct ceph_connection *con)
+{
+	int ret;
+
+	dout("%s connecting\n", __func__);
+	ret = read_partial_banner(con);
+	if (ret > 0)
+		ret = process_banner(con);
+
+	return ret;
+}
 
 /*
  * Write something to the socket.  Called in a worker thread when the

@@ -1986,17 +2037,7 @@ more:
 
 	/* open the socket first? */
 	if (con->sock == NULL) {
-		set_bit(CONNECTING, &con->state);
-
-		con_out_kvec_reset(con);
-		prepare_write_banner(con);
-		prepare_read_banner(con);
-
-		BUG_ON(con->in_msg);
-		con->in_tag = CEPH_MSGR_TAG_READY;
-		dout("try_write initiating connect on %p new state %lu\n",
-		     con, con->state);
-		ret = ceph_tcp_connect(con);
+		ret = ceph_con_connect(con);
 		if (ret < 0) {
 			con->error_msg = "connect error";
 			goto out;

@@ -2095,13 +2136,9 @@ more:
 	}
 
 	if (test_bit(CONNECTING, &con->state)) {
-		dout("try_read connecting\n");
-		ret = read_partial_banner(con);
+		ret = ceph_con_connect_response(con);
 		if (ret <= 0)
 			goto out;
-		ret = process_banner(con);
-		if (ret < 0)
-			goto out;
 
 		clear_bit(CONNECTING, &con->state);
 		set_bit(NEGOTIATING, &con->state);

Index: b/net/ceph/messenger.c
===================================================================
@@ -1633,7 +1633,7 @@ static int process_connect(struct ceph_c
 		con->error_msg = "protocol error, garbage tag during connect";
 		return -1;
 	}
-	return 0;
+	return 1;
 }
 
 

@@ -2023,6 +2023,82 @@ static int ceph_con_connect_response(str
 }
 
 /*
+ * The first phase of connecting with the peer succeeded.  Now start
+ * the second phase (negotiating), which consists of:
+ *  - client sends a connect message to server, specifying
+ *    information about itself, including the protocol it intends to
+ *    use and the features it supports.
+ *  - if authorizer data is needed for the connection, its length is
+ *    recorded in the connect message, and client sends its content
+ *    immediately after the connect message
+ *  - server receives the connect message from the client, and if it
+ *    indicates authorizer data follows, reads that also.
+ * If all is well to this point, then we begin processing the
+ * negotiation response.
+ */
+static int ceph_con_negotiate(struct ceph_connection *con)
+{
+	int ret;
+
+	clear_bit(CONNECTING, &con->state);
+	set_bit(NEGOTIATING, &con->state);
+
+	/* Banner was good, exchange connection info */
+	ret = prepare_write_connect(con);
+	if (ret >= 0)
+		prepare_read_connect(con);
+
+	return ret;
+}
+
+/*
+ * Handle the response from the negotiating phase of connecting the
+ * peer.  This consists of:
+ *  - server validates the connect message (and possibly authorizer
+ *    data), and sends a response to the client:
+ *      - if the protocol version supplied by the client is not what
+ *        was expected, response is a BADPROTOVER tag
+ *      - if the features supported by the client are missing
+ *        features required by the server, response is a FEATURES
+ *        tag.
+ *      - if the features supported by the client are missing
+ *      - if authorizer data is supplied by the client and it is not
+ *        valid, response is a BADAUTHORIZER tag.
+ *      - (There are some other conditions related to message and
+ *        connection sequence numbers but they are not covered here)
+ *      - Otherwise the response will begin with a READY tag, and
+ *        will include a ceph connect reply message, which will
+ *        include the features supported by the server, and the
+ *        server's own authorization data.
+ *  - client validates the connect message (and possibly authorizer
+ *    data) from the server:
+ *      - If the tag indicates a bad protocol or mismatching
+ *        features, the connection attempt is abandoned, so the ceph
+ *        connection is reset and closed.
+ *      - If the tag indicates a bad authorizer, a second connect
+ *        attempt is initiated.  If a second attempt fails due to a
+ *        bad authorizer, the connection attempt fails.
+ *      - If the tag indicates READY, the client will check the
+ *        features supported by the server.  If the server's
+ *        features do not include a feature required by the client,
+ *        the connection attempt is abandoned, so the ceph
+ *        connection is reset and closed.
+ *  If no failures occurred to this point, the connection is established.
+ */
+static int ceph_con_negotiate_response(struct ceph_connection *con)
+{
+	int ret;
+
+	dout("%s negotiating\n", __func__);
+
+	ret = read_partial_connect(con);
+	if (ret > 0)
+		ret = process_connect(con);
+
+	return ret;
+}
+
+/*
  * Write something to the socket.  Called in a worker thread when the
  * socket appears to be writeable and we have something ready to send.
  */

@@ -2136,31 +2212,30 @@ more:
 	}
 
 	if (test_bit(CONNECTING, &con->state)) {
+		/*
+		 * See if we got the response we expect from our
+		 * connection request.
+		 */
 		ret = ceph_con_connect_response(con);
 		if (ret <= 0)
 			goto out;
 
-		clear_bit(CONNECTING, &con->state);
-		set_bit(NEGOTIATING, &con->state);
-
-		/* Banner is good, exchange connection info */
-		ret = prepare_write_connect(con);
-		if (ret < 0)
-			goto out;
-		prepare_read_connect(con);
+		/*
+		 * All good.  Initiate the negotiation phase of the
+		 * connection.  If this succeeds, we're done reading
+		 * and we next need to send the messages we've
+		 * queued up.  If it fails, we're also done.
+		 */
+		ret = ceph_con_negotiate(con);
 
-		/* Send connection info before awaiting response */
-		goto out;
+		goto out;	/* Regardless of result */
 	}
 
 	if (test_bit(NEGOTIATING, &con->state)) {
-		dout("try_read negotiating\n");
-		ret = read_partial_connect(con);
+		ret = ceph_con_negotiate_response(con);
 		if (ret <= 0)
 			goto out;
-		ret = process_connect(con);
-		if (ret < 0)
-			goto out;
+
 		goto more;
 	}
 

Index: b/net/ceph/messenger.c
===================================================================
@@ -492,6 +492,7 @@ static void reset_connection(struct ceph
 	}
 	con->in_seq = 0;
 	con->in_seq_acked = 0;
+	con_close_socket(con);
 }
 
 /*

Index: b/net/ceph/messenger.c
===================================================================
@@ -2387,7 +2387,6 @@ restart:
 	if (test_and_clear_bit(OPENING, &con->state)) {
 		/* reopen w/ new peer */
 		dout("con_work OPENING\n");
-		con_close_socket(con);
 	}
 
 	ret = try_read(con);

Index: b/net/ceph/messenger.c
===================================================================
@@ -2273,8 +2273,7 @@ more:
 			prepare_read_ack(con);
 			break;
 		case CEPH_MSGR_TAG_CLOSE:
-			clear_bit(CONNECTED, &con->state);
-			set_bit(CLOSED, &con->state);   /* fixme */
+			ret = -EIO;
 			goto out;
 		default:
 			goto bad_tag;

Index: b/net/ceph/messenger.c
===================================================================
@@ -1476,12 +1476,6 @@ static int process_banner(struct ceph_co
 	return 1;
 }
 
-static void fail_protocol(struct ceph_connection *con)
-{
-	reset_connection(con);
-	set_bit(CLOSED, &con->state);  /* in case there's queued work */
-}
-
 static int process_connect(struct ceph_connection *con)
 {
 	u64 sup_feat = con->msgr->supported_features;

@@ -1499,8 +1493,7 @@ static int process_connect(struct ceph_c
 		       ceph_pr_addr(&con->peer_addr.in_addr),
 		       sup_feat, server_feat, server_feat & ~sup_feat);
 		con->error_msg = "missing required protocol features";
-		fail_protocol(con);
-		return -1;
+		return -EIO;
 
 	case CEPH_MSGR_TAG_BADPROTOVER:
 		pr_err("%s%lld %s protocol version mismatch,"

@@ -1510,8 +1503,7 @@ static int process_connect(struct ceph_c
 		       le32_to_cpu(con->out_connect.protocol_version),
 		       le32_to_cpu(con->in_reply.protocol_version));
 		con->error_msg = "protocol version mismatch";
-		fail_protocol(con);
-		return -1;
+		return -EIO;
 
 	case CEPH_MSGR_TAG_BADAUTHORIZER:
 		con->auth_retry++;

@@ -1597,8 +1589,7 @@ static int process_connect(struct ceph_c
 			       ceph_pr_addr(&con->peer_addr.in_addr),
 			       req_feat, server_feat, req_feat & ~server_feat);
 			con->error_msg = "missing required protocol features";
-			fail_protocol(con);
-			return -1;
+			return -EIO;
 		}
 		clear_bit(NEGOTIATING, &con->state);
 		set_bit(CONNECTED, &con->state);

Index: b/net/ceph/messenger.c
===================================================================
@@ -1533,7 +1533,8 @@ static int process_connect(struct ceph_c
 		pr_err("%s%lld %s connection reset\n",
 		       ENTITY_NAME(con->peer_name),
 		       ceph_pr_addr(&con->peer_addr.in_addr));
-		reset_connection(con);
+		ceph_con_close(con);
+
 		ret = prepare_write_connect(con);
 		if (ret < 0)
 			return ret;

Index: b/net/ceph/messenger.c
===================================================================
@@ -502,6 +502,8 @@ void ceph_con_close(struct ceph_connecti
 {
 	dout("con_close %p peer %s\n", con,
 	     ceph_pr_addr(&con->peer_addr.in_addr));
+
+	mutex_lock(&con->mutex);
 	clear_bit(NEGOTIATING, &con->state);
 	clear_bit(CONNECTING, &con->state);
 	clear_bit(CONNECTED, &con->state);

@@ -512,11 +514,13 @@ void ceph_con_close(struct ceph_connecti
 	clear_bit(KEEPALIVE_PENDING, &con->flags);
 	clear_bit(WRITE_PENDING, &con->flags);
 
-	mutex_lock(&con->mutex);
+	/* Clear everything out */
 	reset_connection(con);
 	con->peer_global_seq = 0;
 	cancel_delayed_work(&con->work);
+
 	mutex_unlock(&con->mutex);
+
 	queue_con(con);
 }
 EXPORT_SYMBOL(ceph_con_close);

@@ -2372,7 +2376,6 @@ restart:
 	}
 	if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
 		dout("con_work CLOSED\n");
-		con_close_socket(con);
 		goto done;
 	}
 	if (test_and_clear_bit(OPENING, &con->state)) {

Index: b/net/ceph/messenger.c
===================================================================
@@ -533,6 +533,7 @@ void ceph_con_open(struct ceph_connectio
 	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
 	set_bit(OPENING, &con->state);
 	WARN_ON(!test_and_clear_bit(CLOSED, &con->state));
+	set_bit(CONNECTING, &con->state);
 
 	memcpy(&con->peer_addr, addr, sizeof(*addr));
 	con->delay = 0;      /* reset backoff memory */

@@ -1981,8 +1982,6 @@ static void process_message(struct ceph_
  */
 static int ceph_con_connect(struct ceph_connection *con)
 {
-	set_bit(CONNECTING, &con->state);
-
 	con_out_kvec_reset(con);
 	prepare_write_banner(con);
 	prepare_read_banner(con);