libceph: start tracking connection socket state

Start explicitly keeping track of the state of a ceph connection's socket, separate from the state of the connection itself. Create placeholder functions to encapsulate the state transitions. -------- | NEW* | transient initial state -------- | con_sock_state_init() v ---------- | CLOSED | initialized, but no socket (and no ---------- TCP connection) ^ \ | \ con_sock_state_connecting() | ---------------------- | \ + con_sock_state_closed() \ |\ \ | \ \ | ----------- \ | | CLOSING | socket event; \ | ----------- await close \ | ^ | | | | | + con_sock_state_closing() | | / \ | | / --------------- | | / \ v | / -------------- | / -----------------| CONNECTING | socket created, TCP | | / -------------- connect initiated | | | con_sock_state_connected() | | v ------------- | CONNECTED | TCP connection established ------------- Make the socket state an atomic variable, reinforcing that it's a distinct transtion with no possible "intermediate/both" states. This is almost certainly overkill at this point, though the transitions into CONNECTED and CLOSING state do get called via socket callback (the rest of the transitions occur with the connection mutex held). We can back out the atomicity later.

Alex Elder committed 14 years ago

Commit ce2c8901 parent 928443c2 files · +70 −2Browse files

Unified · Split

2 files

include/linux/ceph/messenger.h +6 −2
net/ceph/messenger.c +64 −0

include/linux/ceph/messenger.h+6−2

@@ -137,14 +137,18 @@ struct ceph_connection { · 136 unmodified lines
137	137	const struct ceph_connection_operations *ops;
138	138
139	139	struct ceph_messenger *msgr;
	140	+
	141	+ atomic_t sock_state;
140	142	struct socket *sock;
	143	+ struct ceph_entity_addr peer_addr; /* peer address */
	144	+ struct ceph_entity_addr peer_addr_for_me;
	145	+
141	146	unsigned long flags;
142	147	unsigned long state;
143	148	const char error_msg; / error message, if any */
144	149
145		− struct ceph_entity_addr peer_addr; /* peer address */
146	150	struct ceph_entity_name peer_name; /* peer name */
147		− struct ceph_entity_addr peer_addr_for_me;
	151	+
148	152	unsigned peer_features;
149	153	u32 connect_seq; /* identify the most recent connection
150	154	attempt for this connection, client */

net/ceph/messenger.c+64−0

@@ -29,6 +29,14 @@ · 28 unmodified lines
29	29	* the sender.
30	30	*/
31	31
	32	+/* State values for ceph_connection->sock_state; NEW is assumed to be 0 */
	33	+
	34	+#define CON_SOCK_STATE_NEW 0 /* -> CLOSED */
	35	+#define CON_SOCK_STATE_CLOSED 1 /* -> CONNECTING */
	36	+#define CON_SOCK_STATE_CONNECTING 2 /* -> CONNECTED or -> CLOSING */
	37	+#define CON_SOCK_STATE_CONNECTED 3 /* -> CLOSING or -> CLOSED */
	38	+#define CON_SOCK_STATE_CLOSING 4 /* -> CLOSED */
	39	+
32	40	/* static tag bytes (protocol control messages) */
33	41	static char tag_msg = CEPH_MSGR_TAG_MSG;
34	42	static char tag_ack = CEPH_MSGR_TAG_ACK;
@@ -147,6 +155,55 @@ void ceph_msgr_flush(void) · 112 unmodified lines
147	155	}
148	156	EXPORT_SYMBOL(ceph_msgr_flush);
149	157
	158	+/* Connection socket state transition functions */
	159	+
	160	+static void con_sock_state_init(struct ceph_connection *con)
	161	+{
	162	+ int old_state;
	163	+
	164	+ old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	165	+ if (WARN_ON(old_state != CON_SOCK_STATE_NEW))
	166	+ printk("%s: unexpected old state %d\n", __func__, old_state);
	167	+}
	168	+
	169	+static void con_sock_state_connecting(struct ceph_connection *con)
	170	+{
	171	+ int old_state;
	172	+
	173	+ old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTING);
	174	+ if (WARN_ON(old_state != CON_SOCK_STATE_CLOSED))
	175	+ printk("%s: unexpected old state %d\n", __func__, old_state);
	176	+}
	177	+
	178	+static void con_sock_state_connected(struct ceph_connection *con)
	179	+{
	180	+ int old_state;
	181	+
	182	+ old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTED);
	183	+ if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING))
	184	+ printk("%s: unexpected old state %d\n", __func__, old_state);
	185	+}
	186	+
	187	+static void con_sock_state_closing(struct ceph_connection *con)
	188	+{
	189	+ int old_state;
	190	+
	191	+ old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSING);
	192	+ if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING &&
	193	+ old_state != CON_SOCK_STATE_CONNECTED &&
	194	+ old_state != CON_SOCK_STATE_CLOSING))
	195	+ printk("%s: unexpected old state %d\n", __func__, old_state);
	196	+}
	197	+
	198	+static void con_sock_state_closed(struct ceph_connection *con)
	199	+{
	200	+ int old_state;
	201	+
	202	+ old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	203	+ if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTED &&
	204	+ old_state != CON_SOCK_STATE_CLOSING))
	205	+ printk("%s: unexpected old state %d\n", __func__, old_state);
	206	+}
150	207
151	208	/*
152	209	* socket callback functions
@@ -203,6 +260,7 @@ static void ceph_sock_state_change(struct sock *sk) · 50 unmodified lines
203	260	dout("%s TCP_CLOSE\n", __func__);
204	261	case TCP_CLOSE_WAIT:
205	262	dout("%s TCP_CLOSE_WAIT\n", __func__);
	263	+ con_sock_state_closing(con);
206	264	if (test_and_set_bit(SOCK_CLOSED, &con->flags) == 0) {
207	265	if (test_bit(CONNECTING, &con->state))
208	266	con->error_msg = "connection failed";
@@ -213,6 +271,7 @@ static void ceph_sock_state_change(struct sock *sk) · 4 unmodified lines
213	271	break;
214	272	case TCP_ESTABLISHED:
215	273	dout("%s TCP_ESTABLISHED\n", __func__);
	274	+ con_sock_state_connected(con);
216	275	queue_con(con);
217	276	break;
218	277	default: /* Everything else is uninteresting */
@@ -277,6 +336,7 @@ static int ceph_tcp_connect(struct ceph_connection *con) · 58 unmodified lines
277	336	return ret;
278	337	}
279	338	con->sock = sock;
	339	+ con_sock_state_connecting(con);
280	340
281	341	return 0;
282	342	}
@@ -343,6 +403,7 @@ static int con_close_socket(struct ceph_connection *con) · 60 unmodified lines
343	403	sock_release(con->sock);
344	404	con->sock = NULL;
345	405	clear_bit(SOCK_CLOSED, &con->state);
	406	+ con_sock_state_closed(con);
346	407	return rc;
347	408	}
348	409
@@ -462,6 +523,9 @@ void ceph_con_init(struct ceph_messenger msgr, struct ceph_connection con) · 113 unmodified lines
462	523	memset(con, 0, sizeof(*con));
463	524	atomic_set(&con->nref, 1);
464	525	con->msgr = msgr;
	526	+
	527	+ con_sock_state_init(con);
	528	+
465	529	mutex_init(&con->mutex);
466	530	INIT_LIST_HEAD(&con->out_queue);
467	531	INIT_LIST_HEAD(&con->out_sent);