1 #include <linux/ceph/ceph_debug.h>
2 
3 #include <linux/crc32c.h>
4 #include <linux/ctype.h>
5 #include <linux/highmem.h>
6 #include <linux/inet.h>
7 #include <linux/kthread.h>
8 #include <linux/net.h>
9 #include <linux/nsproxy.h>
10 #include <linux/slab.h>
11 #include <linux/socket.h>
12 #include <linux/string.h>
13 #ifdef	CONFIG_BLOCK
14 #include <linux/bio.h>
15 #endif	/* CONFIG_BLOCK */
16 #include <linux/dns_resolver.h>
17 #include <net/tcp.h>
18 
19 #include <linux/ceph/ceph_features.h>
20 #include <linux/ceph/libceph.h>
21 #include <linux/ceph/messenger.h>
22 #include <linux/ceph/decode.h>
23 #include <linux/ceph/pagelist.h>
24 #include <linux/export.h>
25 
26 #define list_entry_next(pos, member)					\
27 	list_entry(pos->member.next, typeof(*pos), member)
28 
29 /*
30  * Ceph uses the messenger to exchange ceph_msg messages with other
31  * hosts in the system.  The messenger provides ordered and reliable
32  * delivery.  We tolerate TCP disconnects by reconnecting (with
33  * exponential backoff) in the case of a fault (disconnection, bad
34  * crc, protocol error).  Acks allow sent messages to be discarded by
35  * the sender.
36  */
37 
38 /*
39  * We track the state of the socket on a given connection using
40  * values defined below.  The transition to a new socket state is
41  * handled by a function which verifies we aren't coming from an
42  * unexpected state.
43  *
44  *      --------
45  *      | NEW* |  transient initial state
46  *      --------
47  *          | con_sock_state_init()
48  *          v
49  *      ----------
50  *      | CLOSED |  initialized, but no socket (and no
51  *      ----------  TCP connection)
52  *       ^      \
53  *       |       \ con_sock_state_connecting()
54  *       |        ----------------------
55  *       |                              \
56  *       + con_sock_state_closed()       \
57  *       |+---------------------------    \
58  *       | \                          \    \
59  *       |  -----------                \    \
60  *       |  | CLOSING |  socket event;  \    \
61  *       |  -----------  await close     \    \
62  *       |       ^                        \   |
63  *       |       |                         \  |
64  *       |       + con_sock_state_closing() \ |
65  *       |      / \                         | |
66  *       |     /   ---------------          | |
67  *       |    /                   \         v v
68  *       |   /                    --------------
69  *       |  /    -----------------| CONNECTING |  socket created, TCP
70  *       |  |   /                 --------------  connect initiated
71  *       |  |   | con_sock_state_connected()
72  *       |  |   v
73  *      -------------
74  *      | CONNECTED |  TCP connection established
75  *      -------------
76  *
77  * State values for ceph_connection->sock_state; NEW is assumed to be 0.
78  */
79 
80 #define CON_SOCK_STATE_NEW		0	/* -> CLOSED */
81 #define CON_SOCK_STATE_CLOSED		1	/* -> CONNECTING */
82 #define CON_SOCK_STATE_CONNECTING	2	/* -> CONNECTED or -> CLOSING */
83 #define CON_SOCK_STATE_CONNECTED	3	/* -> CLOSING or -> CLOSED */
84 #define CON_SOCK_STATE_CLOSING		4	/* -> CLOSED */
85 
86 /*
87  * connection states
88  */
89 #define CON_STATE_CLOSED        1  /* -> PREOPEN */
90 #define CON_STATE_PREOPEN       2  /* -> CONNECTING, CLOSED */
91 #define CON_STATE_CONNECTING    3  /* -> NEGOTIATING, CLOSED */
92 #define CON_STATE_NEGOTIATING   4  /* -> OPEN, CLOSED */
93 #define CON_STATE_OPEN          5  /* -> STANDBY, CLOSED */
94 #define CON_STATE_STANDBY       6  /* -> PREOPEN, CLOSED */
95 
96 /*
97  * ceph_connection flag bits
98  */
99 #define CON_FLAG_LOSSYTX           0  /* we can close channel or drop
100 				       * messages on errors */
101 #define CON_FLAG_KEEPALIVE_PENDING 1  /* we need to send a keepalive */
102 #define CON_FLAG_WRITE_PENDING	   2  /* we have data ready to send */
103 #define CON_FLAG_SOCK_CLOSED	   3  /* socket state changed to closed */
104 #define CON_FLAG_BACKOFF           4  /* need to retry queuing delayed work */
105 
con_flag_valid(unsigned long con_flag)106 static bool con_flag_valid(unsigned long con_flag)
107 {
108 	switch (con_flag) {
109 	case CON_FLAG_LOSSYTX:
110 	case CON_FLAG_KEEPALIVE_PENDING:
111 	case CON_FLAG_WRITE_PENDING:
112 	case CON_FLAG_SOCK_CLOSED:
113 	case CON_FLAG_BACKOFF:
114 		return true;
115 	default:
116 		return false;
117 	}
118 }
119 
con_flag_clear(struct ceph_connection * con,unsigned long con_flag)120 static void con_flag_clear(struct ceph_connection *con, unsigned long con_flag)
121 {
122 	BUG_ON(!con_flag_valid(con_flag));
123 
124 	clear_bit(con_flag, &con->flags);
125 }
126 
con_flag_set(struct ceph_connection * con,unsigned long con_flag)127 static void con_flag_set(struct ceph_connection *con, unsigned long con_flag)
128 {
129 	BUG_ON(!con_flag_valid(con_flag));
130 
131 	set_bit(con_flag, &con->flags);
132 }
133 
con_flag_test(struct ceph_connection * con,unsigned long con_flag)134 static bool con_flag_test(struct ceph_connection *con, unsigned long con_flag)
135 {
136 	BUG_ON(!con_flag_valid(con_flag));
137 
138 	return test_bit(con_flag, &con->flags);
139 }
140 
con_flag_test_and_clear(struct ceph_connection * con,unsigned long con_flag)141 static bool con_flag_test_and_clear(struct ceph_connection *con,
142 					unsigned long con_flag)
143 {
144 	BUG_ON(!con_flag_valid(con_flag));
145 
146 	return test_and_clear_bit(con_flag, &con->flags);
147 }
148 
con_flag_test_and_set(struct ceph_connection * con,unsigned long con_flag)149 static bool con_flag_test_and_set(struct ceph_connection *con,
150 					unsigned long con_flag)
151 {
152 	BUG_ON(!con_flag_valid(con_flag));
153 
154 	return test_and_set_bit(con_flag, &con->flags);
155 }
156 
157 /* Slab caches for frequently-allocated structures */
158 
159 static struct kmem_cache	*ceph_msg_cache;
160 static struct kmem_cache	*ceph_msg_data_cache;
161 
162 /* static tag bytes (protocol control messages) */
163 static char tag_msg = CEPH_MSGR_TAG_MSG;
164 static char tag_ack = CEPH_MSGR_TAG_ACK;
165 static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
166 static char tag_keepalive2 = CEPH_MSGR_TAG_KEEPALIVE2;
167 
168 #ifdef CONFIG_LOCKDEP
169 static struct lock_class_key socket_class;
170 #endif
171 
172 /*
173  * When skipping (ignoring) a block of input we read it into a "skip
174  * buffer," which is this many bytes in size.
175  */
176 #define SKIP_BUF_SIZE	1024
177 
178 static void queue_con(struct ceph_connection *con);
179 static void cancel_con(struct ceph_connection *con);
180 static void ceph_con_workfn(struct work_struct *);
181 static void con_fault(struct ceph_connection *con);
182 
183 /*
184  * Nicely render a sockaddr as a string.  An array of formatted
185  * strings is used, to approximate reentrancy.
186  */
187 #define ADDR_STR_COUNT_LOG	5	/* log2(# address strings in array) */
188 #define ADDR_STR_COUNT		(1 << ADDR_STR_COUNT_LOG)
189 #define ADDR_STR_COUNT_MASK	(ADDR_STR_COUNT - 1)
190 #define MAX_ADDR_STR_LEN	64	/* 54 is enough */
191 
192 static char addr_str[ADDR_STR_COUNT][MAX_ADDR_STR_LEN];
193 static atomic_t addr_str_seq = ATOMIC_INIT(0);
194 
195 static struct page *zero_page;		/* used in certain error cases */
196 
ceph_pr_addr(const struct sockaddr_storage * ss)197 const char *ceph_pr_addr(const struct sockaddr_storage *ss)
198 {
199 	int i;
200 	char *s;
201 	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
202 	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
203 
204 	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
205 	s = addr_str[i];
206 
207 	switch (ss->ss_family) {
208 	case AF_INET:
209 		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
210 			 ntohs(in4->sin_port));
211 		break;
212 
213 	case AF_INET6:
214 		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
215 			 ntohs(in6->sin6_port));
216 		break;
217 
218 	default:
219 		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
220 			 ss->ss_family);
221 	}
222 
223 	return s;
224 }
225 EXPORT_SYMBOL(ceph_pr_addr);
226 
encode_my_addr(struct ceph_messenger * msgr)227 static void encode_my_addr(struct ceph_messenger *msgr)
228 {
229 	memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
230 	ceph_encode_addr(&msgr->my_enc_addr);
231 }
232 
233 /*
234  * work queue for all reading and writing to/from the socket.
235  */
236 static struct workqueue_struct *ceph_msgr_wq;
237 
ceph_msgr_slab_init(void)238 static int ceph_msgr_slab_init(void)
239 {
240 	BUG_ON(ceph_msg_cache);
241 	ceph_msg_cache = kmem_cache_create("ceph_msg",
242 					sizeof (struct ceph_msg),
243 					__alignof__(struct ceph_msg), 0, NULL);
244 
245 	if (!ceph_msg_cache)
246 		return -ENOMEM;
247 
248 	BUG_ON(ceph_msg_data_cache);
249 	ceph_msg_data_cache = kmem_cache_create("ceph_msg_data",
250 					sizeof (struct ceph_msg_data),
251 					__alignof__(struct ceph_msg_data),
252 					0, NULL);
253 	if (ceph_msg_data_cache)
254 		return 0;
255 
256 	kmem_cache_destroy(ceph_msg_cache);
257 	ceph_msg_cache = NULL;
258 
259 	return -ENOMEM;
260 }
261 
ceph_msgr_slab_exit(void)262 static void ceph_msgr_slab_exit(void)
263 {
264 	BUG_ON(!ceph_msg_data_cache);
265 	kmem_cache_destroy(ceph_msg_data_cache);
266 	ceph_msg_data_cache = NULL;
267 
268 	BUG_ON(!ceph_msg_cache);
269 	kmem_cache_destroy(ceph_msg_cache);
270 	ceph_msg_cache = NULL;
271 }
272 
_ceph_msgr_exit(void)273 static void _ceph_msgr_exit(void)
274 {
275 	if (ceph_msgr_wq) {
276 		destroy_workqueue(ceph_msgr_wq);
277 		ceph_msgr_wq = NULL;
278 	}
279 
280 	BUG_ON(zero_page == NULL);
281 	page_cache_release(zero_page);
282 	zero_page = NULL;
283 
284 	ceph_msgr_slab_exit();
285 }
286 
ceph_msgr_init(void)287 int ceph_msgr_init(void)
288 {
289 	if (ceph_msgr_slab_init())
290 		return -ENOMEM;
291 
292 	BUG_ON(zero_page != NULL);
293 	zero_page = ZERO_PAGE(0);
294 	page_cache_get(zero_page);
295 
296 	/*
297 	 * The number of active work items is limited by the number of
298 	 * connections, so leave @max_active at default.
299 	 */
300 	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_MEM_RECLAIM, 0);
301 	if (ceph_msgr_wq)
302 		return 0;
303 
304 	pr_err("msgr_init failed to create workqueue\n");
305 	_ceph_msgr_exit();
306 
307 	return -ENOMEM;
308 }
309 EXPORT_SYMBOL(ceph_msgr_init);
310 
ceph_msgr_exit(void)311 void ceph_msgr_exit(void)
312 {
313 	BUG_ON(ceph_msgr_wq == NULL);
314 
315 	_ceph_msgr_exit();
316 }
317 EXPORT_SYMBOL(ceph_msgr_exit);
318 
ceph_msgr_flush(void)319 void ceph_msgr_flush(void)
320 {
321 	flush_workqueue(ceph_msgr_wq);
322 }
323 EXPORT_SYMBOL(ceph_msgr_flush);
324 
325 /* Connection socket state transition functions */
326 
con_sock_state_init(struct ceph_connection * con)327 static void con_sock_state_init(struct ceph_connection *con)
328 {
329 	int old_state;
330 
331 	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
332 	if (WARN_ON(old_state != CON_SOCK_STATE_NEW))
333 		printk("%s: unexpected old state %d\n", __func__, old_state);
334 	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
335 	     CON_SOCK_STATE_CLOSED);
336 }
337 
con_sock_state_connecting(struct ceph_connection * con)338 static void con_sock_state_connecting(struct ceph_connection *con)
339 {
340 	int old_state;
341 
342 	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTING);
343 	if (WARN_ON(old_state != CON_SOCK_STATE_CLOSED))
344 		printk("%s: unexpected old state %d\n", __func__, old_state);
345 	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
346 	     CON_SOCK_STATE_CONNECTING);
347 }
348 
con_sock_state_connected(struct ceph_connection * con)349 static void con_sock_state_connected(struct ceph_connection *con)
350 {
351 	int old_state;
352 
353 	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTED);
354 	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING))
355 		printk("%s: unexpected old state %d\n", __func__, old_state);
356 	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
357 	     CON_SOCK_STATE_CONNECTED);
358 }
359 
con_sock_state_closing(struct ceph_connection * con)360 static void con_sock_state_closing(struct ceph_connection *con)
361 {
362 	int old_state;
363 
364 	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSING);
365 	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING &&
366 			old_state != CON_SOCK_STATE_CONNECTED &&
367 			old_state != CON_SOCK_STATE_CLOSING))
368 		printk("%s: unexpected old state %d\n", __func__, old_state);
369 	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
370 	     CON_SOCK_STATE_CLOSING);
371 }
372 
con_sock_state_closed(struct ceph_connection * con)373 static void con_sock_state_closed(struct ceph_connection *con)
374 {
375 	int old_state;
376 
377 	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
378 	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTED &&
379 		    old_state != CON_SOCK_STATE_CLOSING &&
380 		    old_state != CON_SOCK_STATE_CONNECTING &&
381 		    old_state != CON_SOCK_STATE_CLOSED))
382 		printk("%s: unexpected old state %d\n", __func__, old_state);
383 	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
384 	     CON_SOCK_STATE_CLOSED);
385 }
386 
387 /*
388  * socket callback functions
389  */
390 
391 /* data available on socket, or listen socket received a connect */
ceph_sock_data_ready(struct sock * sk)392 static void ceph_sock_data_ready(struct sock *sk)
393 {
394 	struct ceph_connection *con = sk->sk_user_data;
395 	if (atomic_read(&con->msgr->stopping)) {
396 		return;
397 	}
398 
399 	if (sk->sk_state != TCP_CLOSE_WAIT) {
400 		dout("%s on %p state = %lu, queueing work\n", __func__,
401 		     con, con->state);
402 		queue_con(con);
403 	}
404 }
405 
406 /* socket has buffer space for writing */
ceph_sock_write_space(struct sock * sk)407 static void ceph_sock_write_space(struct sock *sk)
408 {
409 	struct ceph_connection *con = sk->sk_user_data;
410 
411 	/* only queue to workqueue if there is data we want to write,
412 	 * and there is sufficient space in the socket buffer to accept
413 	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
414 	 * doesn't get called again until try_write() fills the socket
415 	 * buffer. See net/ipv4/tcp_input.c:tcp_check_space()
416 	 * and net/core/stream.c:sk_stream_write_space().
417 	 */
418 	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
419 		if (sk_stream_is_writeable(sk)) {
420 			dout("%s %p queueing write work\n", __func__, con);
421 			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
422 			queue_con(con);
423 		}
424 	} else {
425 		dout("%s %p nothing to write\n", __func__, con);
426 	}
427 }
428 
429 /* socket's state has changed */
ceph_sock_state_change(struct sock * sk)430 static void ceph_sock_state_change(struct sock *sk)
431 {
432 	struct ceph_connection *con = sk->sk_user_data;
433 
434 	dout("%s %p state = %lu sk_state = %u\n", __func__,
435 	     con, con->state, sk->sk_state);
436 
437 	switch (sk->sk_state) {
438 	case TCP_CLOSE:
439 		dout("%s TCP_CLOSE\n", __func__);
440 	case TCP_CLOSE_WAIT:
441 		dout("%s TCP_CLOSE_WAIT\n", __func__);
442 		con_sock_state_closing(con);
443 		con_flag_set(con, CON_FLAG_SOCK_CLOSED);
444 		queue_con(con);
445 		break;
446 	case TCP_ESTABLISHED:
447 		dout("%s TCP_ESTABLISHED\n", __func__);
448 		con_sock_state_connected(con);
449 		queue_con(con);
450 		break;
451 	default:	/* Everything else is uninteresting */
452 		break;
453 	}
454 }
455 
456 /*
457  * set up socket callbacks
458  */
set_sock_callbacks(struct socket * sock,struct ceph_connection * con)459 static void set_sock_callbacks(struct socket *sock,
460 			       struct ceph_connection *con)
461 {
462 	struct sock *sk = sock->sk;
463 	sk->sk_user_data = con;
464 	sk->sk_data_ready = ceph_sock_data_ready;
465 	sk->sk_write_space = ceph_sock_write_space;
466 	sk->sk_state_change = ceph_sock_state_change;
467 }
468 
469 
470 /*
471  * socket helpers
472  */
473 
474 /*
475  * initiate connection to a remote socket.
476  */
ceph_tcp_connect(struct ceph_connection * con)477 static int ceph_tcp_connect(struct ceph_connection *con)
478 {
479 	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
480 	struct socket *sock;
481 	int ret;
482 
483 	BUG_ON(con->sock);
484 	ret = sock_create_kern(read_pnet(&con->msgr->net), paddr->ss_family,
485 			       SOCK_STREAM, IPPROTO_TCP, &sock);
486 	if (ret)
487 		return ret;
488 	sock->sk->sk_allocation = GFP_NOFS;
489 
490 #ifdef CONFIG_LOCKDEP
491 	lockdep_set_class(&sock->sk->sk_lock, &socket_class);
492 #endif
493 
494 	set_sock_callbacks(sock, con);
495 
496 	dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
497 
498 	con_sock_state_connecting(con);
499 	ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
500 				 O_NONBLOCK);
501 	if (ret == -EINPROGRESS) {
502 		dout("connect %s EINPROGRESS sk_state = %u\n",
503 		     ceph_pr_addr(&con->peer_addr.in_addr),
504 		     sock->sk->sk_state);
505 	} else if (ret < 0) {
506 		pr_err("connect %s error %d\n",
507 		       ceph_pr_addr(&con->peer_addr.in_addr), ret);
508 		sock_release(sock);
509 		return ret;
510 	}
511 
512 	if (ceph_test_opt(from_msgr(con->msgr), TCP_NODELAY)) {
513 		int optval = 1;
514 
515 		ret = kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
516 					(char *)&optval, sizeof(optval));
517 		if (ret)
518 			pr_err("kernel_setsockopt(TCP_NODELAY) failed: %d",
519 			       ret);
520 	}
521 
522 	con->sock = sock;
523 	return 0;
524 }
525 
ceph_tcp_recvmsg(struct socket * sock,void * buf,size_t len)526 static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
527 {
528 	struct kvec iov = {buf, len};
529 	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
530 	int r;
531 
532 	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
533 	if (r == -EAGAIN)
534 		r = 0;
535 	return r;
536 }
537 
ceph_tcp_recvpage(struct socket * sock,struct page * page,int page_offset,size_t length)538 static int ceph_tcp_recvpage(struct socket *sock, struct page *page,
539 		     int page_offset, size_t length)
540 {
541 	void *kaddr;
542 	int ret;
543 
544 	BUG_ON(page_offset + length > PAGE_SIZE);
545 
546 	kaddr = kmap(page);
547 	BUG_ON(!kaddr);
548 	ret = ceph_tcp_recvmsg(sock, kaddr + page_offset, length);
549 	kunmap(page);
550 
551 	return ret;
552 }
553 
554 /*
555  * write something.  @more is true if caller will be sending more data
556  * shortly.
557  */
ceph_tcp_sendmsg(struct socket * sock,struct kvec * iov,size_t kvlen,size_t len,int more)558 static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
559 		     size_t kvlen, size_t len, int more)
560 {
561 	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
562 	int r;
563 
564 	if (more)
565 		msg.msg_flags |= MSG_MORE;
566 	else
567 		msg.msg_flags |= MSG_EOR;  /* superfluous, but what the hell */
568 
569 	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
570 	if (r == -EAGAIN)
571 		r = 0;
572 	return r;
573 }
574 
__ceph_tcp_sendpage(struct socket * sock,struct page * page,int offset,size_t size,bool more)575 static int __ceph_tcp_sendpage(struct socket *sock, struct page *page,
576 		     int offset, size_t size, bool more)
577 {
578 	int flags = MSG_DONTWAIT | MSG_NOSIGNAL | (more ? MSG_MORE : MSG_EOR);
579 	int ret;
580 
581 	ret = kernel_sendpage(sock, page, offset, size, flags);
582 	if (ret == -EAGAIN)
583 		ret = 0;
584 
585 	return ret;
586 }
587 
ceph_tcp_sendpage(struct socket * sock,struct page * page,int offset,size_t size,bool more)588 static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
589 		     int offset, size_t size, bool more)
590 {
591 	int ret;
592 	struct kvec iov;
593 
594 	/* sendpage cannot properly handle pages with page_count == 0,
595 	 * we need to fallback to sendmsg if that's the case */
596 	if (page_count(page) >= 1)
597 		return __ceph_tcp_sendpage(sock, page, offset, size, more);
598 
599 	iov.iov_base = kmap(page) + offset;
600 	iov.iov_len = size;
601 	ret = ceph_tcp_sendmsg(sock, &iov, 1, size, more);
602 	kunmap(page);
603 
604 	return ret;
605 }
606 
607 /*
608  * Shutdown/close the socket for the given connection.
609  */
con_close_socket(struct ceph_connection * con)610 static int con_close_socket(struct ceph_connection *con)
611 {
612 	int rc = 0;
613 
614 	dout("con_close_socket on %p sock %p\n", con, con->sock);
615 	if (con->sock) {
616 		rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
617 		sock_release(con->sock);
618 		con->sock = NULL;
619 	}
620 
621 	/*
622 	 * Forcibly clear the SOCK_CLOSED flag.  It gets set
623 	 * independent of the connection mutex, and we could have
624 	 * received a socket close event before we had the chance to
625 	 * shut the socket down.
626 	 */
627 	con_flag_clear(con, CON_FLAG_SOCK_CLOSED);
628 
629 	con_sock_state_closed(con);
630 	return rc;
631 }
632 
633 /*
634  * Reset a connection.  Discard all incoming and outgoing messages
635  * and clear *_seq state.
636  */
ceph_msg_remove(struct ceph_msg * msg)637 static void ceph_msg_remove(struct ceph_msg *msg)
638 {
639 	list_del_init(&msg->list_head);
640 
641 	ceph_msg_put(msg);
642 }
ceph_msg_remove_list(struct list_head * head)643 static void ceph_msg_remove_list(struct list_head *head)
644 {
645 	while (!list_empty(head)) {
646 		struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
647 							list_head);
648 		ceph_msg_remove(msg);
649 	}
650 }
651 
reset_connection(struct ceph_connection * con)652 static void reset_connection(struct ceph_connection *con)
653 {
654 	/* reset connection, out_queue, msg_ and connect_seq */
655 	/* discard existing out_queue and msg_seq */
656 	dout("reset_connection %p\n", con);
657 	ceph_msg_remove_list(&con->out_queue);
658 	ceph_msg_remove_list(&con->out_sent);
659 
660 	if (con->in_msg) {
661 		BUG_ON(con->in_msg->con != con);
662 		ceph_msg_put(con->in_msg);
663 		con->in_msg = NULL;
664 	}
665 
666 	con->connect_seq = 0;
667 	con->out_seq = 0;
668 	if (con->out_msg) {
669 		BUG_ON(con->out_msg->con != con);
670 		ceph_msg_put(con->out_msg);
671 		con->out_msg = NULL;
672 	}
673 	con->in_seq = 0;
674 	con->in_seq_acked = 0;
675 
676 	con->out_skip = 0;
677 }
678 
679 /*
680  * mark a peer down.  drop any open connections.
681  */
ceph_con_close(struct ceph_connection * con)682 void ceph_con_close(struct ceph_connection *con)
683 {
684 	mutex_lock(&con->mutex);
685 	dout("con_close %p peer %s\n", con,
686 	     ceph_pr_addr(&con->peer_addr.in_addr));
687 	con->state = CON_STATE_CLOSED;
688 
689 	con_flag_clear(con, CON_FLAG_LOSSYTX);	/* so we retry next connect */
690 	con_flag_clear(con, CON_FLAG_KEEPALIVE_PENDING);
691 	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
692 	con_flag_clear(con, CON_FLAG_BACKOFF);
693 
694 	reset_connection(con);
695 	con->peer_global_seq = 0;
696 	cancel_con(con);
697 	con_close_socket(con);
698 	mutex_unlock(&con->mutex);
699 }
700 EXPORT_SYMBOL(ceph_con_close);
701 
702 /*
703  * Reopen a closed connection, with a new peer address.
704  */
ceph_con_open(struct ceph_connection * con,__u8 entity_type,__u64 entity_num,struct ceph_entity_addr * addr)705 void ceph_con_open(struct ceph_connection *con,
706 		   __u8 entity_type, __u64 entity_num,
707 		   struct ceph_entity_addr *addr)
708 {
709 	mutex_lock(&con->mutex);
710 	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
711 
712 	WARN_ON(con->state != CON_STATE_CLOSED);
713 	con->state = CON_STATE_PREOPEN;
714 
715 	con->peer_name.type = (__u8) entity_type;
716 	con->peer_name.num = cpu_to_le64(entity_num);
717 
718 	memcpy(&con->peer_addr, addr, sizeof(*addr));
719 	con->delay = 0;      /* reset backoff memory */
720 	mutex_unlock(&con->mutex);
721 	queue_con(con);
722 }
723 EXPORT_SYMBOL(ceph_con_open);
724 
725 /*
726  * return true if this connection ever successfully opened
727  */
ceph_con_opened(struct ceph_connection * con)728 bool ceph_con_opened(struct ceph_connection *con)
729 {
730 	return con->connect_seq > 0;
731 }
732 
733 /*
734  * initialize a new connection.
735  */
ceph_con_init(struct ceph_connection * con,void * private,const struct ceph_connection_operations * ops,struct ceph_messenger * msgr)736 void ceph_con_init(struct ceph_connection *con, void *private,
737 	const struct ceph_connection_operations *ops,
738 	struct ceph_messenger *msgr)
739 {
740 	dout("con_init %p\n", con);
741 	memset(con, 0, sizeof(*con));
742 	con->private = private;
743 	con->ops = ops;
744 	con->msgr = msgr;
745 
746 	con_sock_state_init(con);
747 
748 	mutex_init(&con->mutex);
749 	INIT_LIST_HEAD(&con->out_queue);
750 	INIT_LIST_HEAD(&con->out_sent);
751 	INIT_DELAYED_WORK(&con->work, ceph_con_workfn);
752 
753 	con->state = CON_STATE_CLOSED;
754 }
755 EXPORT_SYMBOL(ceph_con_init);
756 
757 
758 /*
759  * We maintain a global counter to order connection attempts.  Get
760  * a unique seq greater than @gt.
761  */
get_global_seq(struct ceph_messenger * msgr,u32 gt)762 static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
763 {
764 	u32 ret;
765 
766 	spin_lock(&msgr->global_seq_lock);
767 	if (msgr->global_seq < gt)
768 		msgr->global_seq = gt;
769 	ret = ++msgr->global_seq;
770 	spin_unlock(&msgr->global_seq_lock);
771 	return ret;
772 }
773 
con_out_kvec_reset(struct ceph_connection * con)774 static void con_out_kvec_reset(struct ceph_connection *con)
775 {
776 	BUG_ON(con->out_skip);
777 
778 	con->out_kvec_left = 0;
779 	con->out_kvec_bytes = 0;
780 	con->out_kvec_cur = &con->out_kvec[0];
781 }
782 
con_out_kvec_add(struct ceph_connection * con,size_t size,void * data)783 static void con_out_kvec_add(struct ceph_connection *con,
784 				size_t size, void *data)
785 {
786 	int index = con->out_kvec_left;
787 
788 	BUG_ON(con->out_skip);
789 	BUG_ON(index >= ARRAY_SIZE(con->out_kvec));
790 
791 	con->out_kvec[index].iov_len = size;
792 	con->out_kvec[index].iov_base = data;
793 	con->out_kvec_left++;
794 	con->out_kvec_bytes += size;
795 }
796 
797 /*
798  * Chop off a kvec from the end.  Return residual number of bytes for
799  * that kvec, i.e. how many bytes would have been written if the kvec
800  * hadn't been nuked.
801  */
con_out_kvec_skip(struct ceph_connection * con)802 static int con_out_kvec_skip(struct ceph_connection *con)
803 {
804 	int off = con->out_kvec_cur - con->out_kvec;
805 	int skip = 0;
806 
807 	if (con->out_kvec_bytes > 0) {
808 		skip = con->out_kvec[off + con->out_kvec_left - 1].iov_len;
809 		BUG_ON(con->out_kvec_bytes < skip);
810 		BUG_ON(!con->out_kvec_left);
811 		con->out_kvec_bytes -= skip;
812 		con->out_kvec_left--;
813 	}
814 
815 	return skip;
816 }
817 
818 #ifdef CONFIG_BLOCK
819 
820 /*
821  * For a bio data item, a piece is whatever remains of the next
822  * entry in the current bio iovec, or the first entry in the next
823  * bio in the list.
824  */
ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor * cursor,size_t length)825 static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
826 					size_t length)
827 {
828 	struct ceph_msg_data *data = cursor->data;
829 	struct bio *bio;
830 
831 	BUG_ON(data->type != CEPH_MSG_DATA_BIO);
832 
833 	bio = data->bio;
834 	BUG_ON(!bio);
835 
836 	cursor->resid = min(length, data->bio_length);
837 	cursor->bio = bio;
838 	cursor->bvec_iter = bio->bi_iter;
839 	cursor->last_piece =
840 		cursor->resid <= bio_iter_len(bio, cursor->bvec_iter);
841 }
842 
ceph_msg_data_bio_next(struct ceph_msg_data_cursor * cursor,size_t * page_offset,size_t * length)843 static struct page *ceph_msg_data_bio_next(struct ceph_msg_data_cursor *cursor,
844 						size_t *page_offset,
845 						size_t *length)
846 {
847 	struct ceph_msg_data *data = cursor->data;
848 	struct bio *bio;
849 	struct bio_vec bio_vec;
850 
851 	BUG_ON(data->type != CEPH_MSG_DATA_BIO);
852 
853 	bio = cursor->bio;
854 	BUG_ON(!bio);
855 
856 	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
857 
858 	*page_offset = (size_t) bio_vec.bv_offset;
859 	BUG_ON(*page_offset >= PAGE_SIZE);
860 	if (cursor->last_piece) /* pagelist offset is always 0 */
861 		*length = cursor->resid;
862 	else
863 		*length = (size_t) bio_vec.bv_len;
864 	BUG_ON(*length > cursor->resid);
865 	BUG_ON(*page_offset + *length > PAGE_SIZE);
866 
867 	return bio_vec.bv_page;
868 }
869 
ceph_msg_data_bio_advance(struct ceph_msg_data_cursor * cursor,size_t bytes)870 static bool ceph_msg_data_bio_advance(struct ceph_msg_data_cursor *cursor,
871 					size_t bytes)
872 {
873 	struct bio *bio;
874 	struct bio_vec bio_vec;
875 
876 	BUG_ON(cursor->data->type != CEPH_MSG_DATA_BIO);
877 
878 	bio = cursor->bio;
879 	BUG_ON(!bio);
880 
881 	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
882 
883 	/* Advance the cursor offset */
884 
885 	BUG_ON(cursor->resid < bytes);
886 	cursor->resid -= bytes;
887 
888 	bio_advance_iter(bio, &cursor->bvec_iter, bytes);
889 
890 	if (bytes < bio_vec.bv_len)
891 		return false;	/* more bytes to process in this segment */
892 
893 	/* Move on to the next segment, and possibly the next bio */
894 
895 	if (!cursor->bvec_iter.bi_size) {
896 		bio = bio->bi_next;
897 		cursor->bio = bio;
898 		if (bio)
899 			cursor->bvec_iter = bio->bi_iter;
900 		else
901 			memset(&cursor->bvec_iter, 0,
902 			       sizeof(cursor->bvec_iter));
903 	}
904 
905 	if (!cursor->last_piece) {
906 		BUG_ON(!cursor->resid);
907 		BUG_ON(!bio);
908 		/* A short read is OK, so use <= rather than == */
909 		if (cursor->resid <= bio_iter_len(bio, cursor->bvec_iter))
910 			cursor->last_piece = true;
911 	}
912 
913 	return true;
914 }
915 #endif /* CONFIG_BLOCK */
916 
917 /*
918  * For a page array, a piece comes from the first page in the array
919  * that has not already been fully consumed.
920  */
ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor * cursor,size_t length)921 static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor,
922 					size_t length)
923 {
924 	struct ceph_msg_data *data = cursor->data;
925 	int page_count;
926 
927 	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
928 
929 	BUG_ON(!data->pages);
930 	BUG_ON(!data->length);
931 
932 	cursor->resid = min(length, data->length);
933 	page_count = calc_pages_for(data->alignment, (u64)data->length);
934 	cursor->page_offset = data->alignment & ~PAGE_MASK;
935 	cursor->page_index = 0;
936 	BUG_ON(page_count > (int)USHRT_MAX);
937 	cursor->page_count = (unsigned short)page_count;
938 	BUG_ON(length > SIZE_MAX - cursor->page_offset);
939 	cursor->last_piece = cursor->page_offset + cursor->resid <= PAGE_SIZE;
940 }
941 
942 static struct page *
ceph_msg_data_pages_next(struct ceph_msg_data_cursor * cursor,size_t * page_offset,size_t * length)943 ceph_msg_data_pages_next(struct ceph_msg_data_cursor *cursor,
944 					size_t *page_offset, size_t *length)
945 {
946 	struct ceph_msg_data *data = cursor->data;
947 
948 	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
949 
950 	BUG_ON(cursor->page_index >= cursor->page_count);
951 	BUG_ON(cursor->page_offset >= PAGE_SIZE);
952 
953 	*page_offset = cursor->page_offset;
954 	if (cursor->last_piece)
955 		*length = cursor->resid;
956 	else
957 		*length = PAGE_SIZE - *page_offset;
958 
959 	return data->pages[cursor->page_index];
960 }
961 
ceph_msg_data_pages_advance(struct ceph_msg_data_cursor * cursor,size_t bytes)962 static bool ceph_msg_data_pages_advance(struct ceph_msg_data_cursor *cursor,
963 						size_t bytes)
964 {
965 	BUG_ON(cursor->data->type != CEPH_MSG_DATA_PAGES);
966 
967 	BUG_ON(cursor->page_offset + bytes > PAGE_SIZE);
968 
969 	/* Advance the cursor page offset */
970 
971 	cursor->resid -= bytes;
972 	cursor->page_offset = (cursor->page_offset + bytes) & ~PAGE_MASK;
973 	if (!bytes || cursor->page_offset)
974 		return false;	/* more bytes to process in the current page */
975 
976 	if (!cursor->resid)
977 		return false;   /* no more data */
978 
979 	/* Move on to the next page; offset is already at 0 */
980 
981 	BUG_ON(cursor->page_index >= cursor->page_count);
982 	cursor->page_index++;
983 	cursor->last_piece = cursor->resid <= PAGE_SIZE;
984 
985 	return true;
986 }
987 
988 /*
989  * For a pagelist, a piece is whatever remains to be consumed in the
990  * first page in the list, or the front of the next page.
991  */
992 static void
ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor * cursor,size_t length)993 ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
994 					size_t length)
995 {
996 	struct ceph_msg_data *data = cursor->data;
997 	struct ceph_pagelist *pagelist;
998 	struct page *page;
999 
1000 	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
1001 
1002 	pagelist = data->pagelist;
1003 	BUG_ON(!pagelist);
1004 
1005 	if (!length)
1006 		return;		/* pagelist can be assigned but empty */
1007 
1008 	BUG_ON(list_empty(&pagelist->head));
1009 	page = list_first_entry(&pagelist->head, struct page, lru);
1010 
1011 	cursor->resid = min(length, pagelist->length);
1012 	cursor->page = page;
1013 	cursor->offset = 0;
1014 	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1015 }
1016 
1017 static struct page *
ceph_msg_data_pagelist_next(struct ceph_msg_data_cursor * cursor,size_t * page_offset,size_t * length)1018 ceph_msg_data_pagelist_next(struct ceph_msg_data_cursor *cursor,
1019 				size_t *page_offset, size_t *length)
1020 {
1021 	struct ceph_msg_data *data = cursor->data;
1022 	struct ceph_pagelist *pagelist;
1023 
1024 	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
1025 
1026 	pagelist = data->pagelist;
1027 	BUG_ON(!pagelist);
1028 
1029 	BUG_ON(!cursor->page);
1030 	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1031 
1032 	/* offset of first page in pagelist is always 0 */
1033 	*page_offset = cursor->offset & ~PAGE_MASK;
1034 	if (cursor->last_piece)
1035 		*length = cursor->resid;
1036 	else
1037 		*length = PAGE_SIZE - *page_offset;
1038 
1039 	return cursor->page;
1040 }
1041 
ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor * cursor,size_t bytes)1042 static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor,
1043 						size_t bytes)
1044 {
1045 	struct ceph_msg_data *data = cursor->data;
1046 	struct ceph_pagelist *pagelist;
1047 
1048 	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
1049 
1050 	pagelist = data->pagelist;
1051 	BUG_ON(!pagelist);
1052 
1053 	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1054 	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);
1055 
1056 	/* Advance the cursor offset */
1057 
1058 	cursor->resid -= bytes;
1059 	cursor->offset += bytes;
1060 	/* offset of first page in pagelist is always 0 */
1061 	if (!bytes || cursor->offset & ~PAGE_MASK)
1062 		return false;	/* more bytes to process in the current page */
1063 
1064 	if (!cursor->resid)
1065 		return false;   /* no more data */
1066 
1067 	/* Move on to the next page */
1068 
1069 	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
1070 	cursor->page = list_entry_next(cursor->page, lru);
1071 	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1072 
1073 	return true;
1074 }
1075 
1076 /*
1077  * Message data is handled (sent or received) in pieces, where each
1078  * piece resides on a single page.  The network layer might not
1079  * consume an entire piece at once.  A data item's cursor keeps
1080  * track of which piece is next to process and how much remains to
1081  * be processed in that piece.  It also tracks whether the current
1082  * piece is the last one in the data item.
1083  */
__ceph_msg_data_cursor_init(struct ceph_msg_data_cursor * cursor)1084 static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
1085 {
1086 	size_t length = cursor->total_resid;
1087 
1088 	switch (cursor->data->type) {
1089 	case CEPH_MSG_DATA_PAGELIST:
1090 		ceph_msg_data_pagelist_cursor_init(cursor, length);
1091 		break;
1092 	case CEPH_MSG_DATA_PAGES:
1093 		ceph_msg_data_pages_cursor_init(cursor, length);
1094 		break;
1095 #ifdef CONFIG_BLOCK
1096 	case CEPH_MSG_DATA_BIO:
1097 		ceph_msg_data_bio_cursor_init(cursor, length);
1098 		break;
1099 #endif /* CONFIG_BLOCK */
1100 	case CEPH_MSG_DATA_NONE:
1101 	default:
1102 		/* BUG(); */
1103 		break;
1104 	}
1105 	cursor->need_crc = true;
1106 }
1107 
ceph_msg_data_cursor_init(struct ceph_msg * msg,size_t length)1108 static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
1109 {
1110 	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1111 	struct ceph_msg_data *data;
1112 
1113 	BUG_ON(!length);
1114 	BUG_ON(length > msg->data_length);
1115 	BUG_ON(list_empty(&msg->data));
1116 
1117 	cursor->data_head = &msg->data;
1118 	cursor->total_resid = length;
1119 	data = list_first_entry(&msg->data, struct ceph_msg_data, links);
1120 	cursor->data = data;
1121 
1122 	__ceph_msg_data_cursor_init(cursor);
1123 }
1124 
1125 /*
1126  * Return the page containing the next piece to process for a given
1127  * data item, and supply the page offset and length of that piece.
1128  * Indicate whether this is the last piece in this data item.
1129  */
ceph_msg_data_next(struct ceph_msg_data_cursor * cursor,size_t * page_offset,size_t * length,bool * last_piece)1130 static struct page *ceph_msg_data_next(struct ceph_msg_data_cursor *cursor,
1131 					size_t *page_offset, size_t *length,
1132 					bool *last_piece)
1133 {
1134 	struct page *page;
1135 
1136 	switch (cursor->data->type) {
1137 	case CEPH_MSG_DATA_PAGELIST:
1138 		page = ceph_msg_data_pagelist_next(cursor, page_offset, length);
1139 		break;
1140 	case CEPH_MSG_DATA_PAGES:
1141 		page = ceph_msg_data_pages_next(cursor, page_offset, length);
1142 		break;
1143 #ifdef CONFIG_BLOCK
1144 	case CEPH_MSG_DATA_BIO:
1145 		page = ceph_msg_data_bio_next(cursor, page_offset, length);
1146 		break;
1147 #endif /* CONFIG_BLOCK */
1148 	case CEPH_MSG_DATA_NONE:
1149 	default:
1150 		page = NULL;
1151 		break;
1152 	}
1153 	BUG_ON(!page);
1154 	BUG_ON(*page_offset + *length > PAGE_SIZE);
1155 	BUG_ON(!*length);
1156 	if (last_piece)
1157 		*last_piece = cursor->last_piece;
1158 
1159 	return page;
1160 }
1161 
1162 /*
1163  * Returns true if the result moves the cursor on to the next piece
1164  * of the data item.
1165  */
ceph_msg_data_advance(struct ceph_msg_data_cursor * cursor,size_t bytes)1166 static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor,
1167 				size_t bytes)
1168 {
1169 	bool new_piece;
1170 
1171 	BUG_ON(bytes > cursor->resid);
1172 	switch (cursor->data->type) {
1173 	case CEPH_MSG_DATA_PAGELIST:
1174 		new_piece = ceph_msg_data_pagelist_advance(cursor, bytes);
1175 		break;
1176 	case CEPH_MSG_DATA_PAGES:
1177 		new_piece = ceph_msg_data_pages_advance(cursor, bytes);
1178 		break;
1179 #ifdef CONFIG_BLOCK
1180 	case CEPH_MSG_DATA_BIO:
1181 		new_piece = ceph_msg_data_bio_advance(cursor, bytes);
1182 		break;
1183 #endif /* CONFIG_BLOCK */
1184 	case CEPH_MSG_DATA_NONE:
1185 	default:
1186 		BUG();
1187 		break;
1188 	}
1189 	cursor->total_resid -= bytes;
1190 
1191 	if (!cursor->resid && cursor->total_resid) {
1192 		WARN_ON(!cursor->last_piece);
1193 		BUG_ON(list_is_last(&cursor->data->links, cursor->data_head));
1194 		cursor->data = list_entry_next(cursor->data, links);
1195 		__ceph_msg_data_cursor_init(cursor);
1196 		new_piece = true;
1197 	}
1198 	cursor->need_crc = new_piece;
1199 
1200 	return new_piece;
1201 }
1202 
sizeof_footer(struct ceph_connection * con)1203 static size_t sizeof_footer(struct ceph_connection *con)
1204 {
1205 	return (con->peer_features & CEPH_FEATURE_MSG_AUTH) ?
1206 	    sizeof(struct ceph_msg_footer) :
1207 	    sizeof(struct ceph_msg_footer_old);
1208 }
1209 
prepare_message_data(struct ceph_msg * msg,u32 data_len)1210 static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
1211 {
1212 	BUG_ON(!msg);
1213 	BUG_ON(!data_len);
1214 
1215 	/* Initialize data cursor */
1216 
1217 	ceph_msg_data_cursor_init(msg, (size_t)data_len);
1218 }
1219 
1220 /*
1221  * Prepare footer for currently outgoing message, and finish things
1222  * off.  Assumes out_kvec* are already valid.. we just add on to the end.
1223  */
prepare_write_message_footer(struct ceph_connection * con)1224 static void prepare_write_message_footer(struct ceph_connection *con)
1225 {
1226 	struct ceph_msg *m = con->out_msg;
1227 	int v = con->out_kvec_left;
1228 
1229 	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;
1230 
1231 	dout("prepare_write_message_footer %p\n", con);
1232 	con->out_kvec[v].iov_base = &m->footer;
1233 	if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
1234 		if (con->ops->sign_message)
1235 			con->ops->sign_message(m);
1236 		else
1237 			m->footer.sig = 0;
1238 		con->out_kvec[v].iov_len = sizeof(m->footer);
1239 		con->out_kvec_bytes += sizeof(m->footer);
1240 	} else {
1241 		m->old_footer.flags = m->footer.flags;
1242 		con->out_kvec[v].iov_len = sizeof(m->old_footer);
1243 		con->out_kvec_bytes += sizeof(m->old_footer);
1244 	}
1245 	con->out_kvec_left++;
1246 	con->out_more = m->more_to_follow;
1247 	con->out_msg_done = true;
1248 }
1249 
1250 /*
1251  * Prepare headers for the next outgoing message.
1252  */
prepare_write_message(struct ceph_connection * con)1253 static void prepare_write_message(struct ceph_connection *con)
1254 {
1255 	struct ceph_msg *m;
1256 	u32 crc;
1257 
1258 	con_out_kvec_reset(con);
1259 	con->out_msg_done = false;
1260 
1261 	/* Sneak an ack in there first?  If we can get it into the same
1262 	 * TCP packet that's a good thing. */
1263 	if (con->in_seq > con->in_seq_acked) {
1264 		con->in_seq_acked = con->in_seq;
1265 		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1266 		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1267 		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1268 			&con->out_temp_ack);
1269 	}
1270 
1271 	BUG_ON(list_empty(&con->out_queue));
1272 	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
1273 	con->out_msg = m;
1274 	BUG_ON(m->con != con);
1275 
1276 	/* put message on sent list */
1277 	ceph_msg_get(m);
1278 	list_move_tail(&m->list_head, &con->out_sent);
1279 
1280 	/*
1281 	 * only assign outgoing seq # if we haven't sent this message
1282 	 * yet.  if it is requeued, resend with it's original seq.
1283 	 */
1284 	if (m->needs_out_seq) {
1285 		m->hdr.seq = cpu_to_le64(++con->out_seq);
1286 		m->needs_out_seq = false;
1287 	}
1288 	WARN_ON(m->data_length != le32_to_cpu(m->hdr.data_len));
1289 
1290 	dout("prepare_write_message %p seq %lld type %d len %d+%d+%zd\n",
1291 	     m, con->out_seq, le16_to_cpu(m->hdr.type),
1292 	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
1293 	     m->data_length);
1294 	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);
1295 
1296 	/* tag + hdr + front + middle */
1297 	con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
1298 	con_out_kvec_add(con, sizeof(con->out_hdr), &con->out_hdr);
1299 	con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
1300 
1301 	if (m->middle)
1302 		con_out_kvec_add(con, m->middle->vec.iov_len,
1303 			m->middle->vec.iov_base);
1304 
1305 	/* fill in hdr crc and finalize hdr */
1306 	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
1307 	con->out_msg->hdr.crc = cpu_to_le32(crc);
1308 	memcpy(&con->out_hdr, &con->out_msg->hdr, sizeof(con->out_hdr));
1309 
1310 	/* fill in front and middle crc, footer */
1311 	crc = crc32c(0, m->front.iov_base, m->front.iov_len);
1312 	con->out_msg->footer.front_crc = cpu_to_le32(crc);
1313 	if (m->middle) {
1314 		crc = crc32c(0, m->middle->vec.iov_base,
1315 				m->middle->vec.iov_len);
1316 		con->out_msg->footer.middle_crc = cpu_to_le32(crc);
1317 	} else
1318 		con->out_msg->footer.middle_crc = 0;
1319 	dout("%s front_crc %u middle_crc %u\n", __func__,
1320 	     le32_to_cpu(con->out_msg->footer.front_crc),
1321 	     le32_to_cpu(con->out_msg->footer.middle_crc));
1322 	con->out_msg->footer.flags = 0;
1323 
1324 	/* is there a data payload? */
1325 	con->out_msg->footer.data_crc = 0;
1326 	if (m->data_length) {
1327 		prepare_message_data(con->out_msg, m->data_length);
1328 		con->out_more = 1;  /* data + footer will follow */
1329 	} else {
1330 		/* no, queue up footer too and be done */
1331 		prepare_write_message_footer(con);
1332 	}
1333 
1334 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1335 }
1336 
1337 /*
1338  * Prepare an ack.
1339  */
prepare_write_ack(struct ceph_connection * con)1340 static void prepare_write_ack(struct ceph_connection *con)
1341 {
1342 	dout("prepare_write_ack %p %llu -> %llu\n", con,
1343 	     con->in_seq_acked, con->in_seq);
1344 	con->in_seq_acked = con->in_seq;
1345 
1346 	con_out_kvec_reset(con);
1347 
1348 	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1349 
1350 	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1351 	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1352 				&con->out_temp_ack);
1353 
1354 	con->out_more = 1;  /* more will follow.. eventually.. */
1355 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1356 }
1357 
1358 /*
1359  * Prepare to share the seq during handshake
1360  */
prepare_write_seq(struct ceph_connection * con)1361 static void prepare_write_seq(struct ceph_connection *con)
1362 {
1363 	dout("prepare_write_seq %p %llu -> %llu\n", con,
1364 	     con->in_seq_acked, con->in_seq);
1365 	con->in_seq_acked = con->in_seq;
1366 
1367 	con_out_kvec_reset(con);
1368 
1369 	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1370 	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1371 			 &con->out_temp_ack);
1372 
1373 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1374 }
1375 
1376 /*
1377  * Prepare to write keepalive byte.
1378  */
prepare_write_keepalive(struct ceph_connection * con)1379 static void prepare_write_keepalive(struct ceph_connection *con)
1380 {
1381 	dout("prepare_write_keepalive %p\n", con);
1382 	con_out_kvec_reset(con);
1383 	if (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2) {
1384 		struct timespec now = CURRENT_TIME;
1385 
1386 		con_out_kvec_add(con, sizeof(tag_keepalive2), &tag_keepalive2);
1387 		ceph_encode_timespec(&con->out_temp_keepalive2, &now);
1388 		con_out_kvec_add(con, sizeof(con->out_temp_keepalive2),
1389 				 &con->out_temp_keepalive2);
1390 	} else {
1391 		con_out_kvec_add(con, sizeof(tag_keepalive), &tag_keepalive);
1392 	}
1393 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1394 }
1395 
1396 /*
1397  * Connection negotiation.
1398  */
1399 
get_connect_authorizer(struct ceph_connection * con,int * auth_proto)1400 static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
1401 						int *auth_proto)
1402 {
1403 	struct ceph_auth_handshake *auth;
1404 
1405 	if (!con->ops->get_authorizer) {
1406 		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
1407 		con->out_connect.authorizer_len = 0;
1408 		return NULL;
1409 	}
1410 
1411 	/* Can't hold the mutex while getting authorizer */
1412 	mutex_unlock(&con->mutex);
1413 	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1414 	mutex_lock(&con->mutex);
1415 
1416 	if (IS_ERR(auth))
1417 		return auth;
1418 	if (con->state != CON_STATE_NEGOTIATING)
1419 		return ERR_PTR(-EAGAIN);
1420 
1421 	con->auth_reply_buf = auth->authorizer_reply_buf;
1422 	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1423 	return auth;
1424 }
1425 
1426 /*
1427  * We connected to a peer and are saying hello.
1428  */
prepare_write_banner(struct ceph_connection * con)1429 static void prepare_write_banner(struct ceph_connection *con)
1430 {
1431 	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
1432 	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1433 					&con->msgr->my_enc_addr);
1434 
1435 	con->out_more = 0;
1436 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1437 }
1438 
prepare_write_connect(struct ceph_connection * con)1439 static int prepare_write_connect(struct ceph_connection *con)
1440 {
1441 	unsigned int global_seq = get_global_seq(con->msgr, 0);
1442 	int proto;
1443 	int auth_proto;
1444 	struct ceph_auth_handshake *auth;
1445 
1446 	switch (con->peer_name.type) {
1447 	case CEPH_ENTITY_TYPE_MON:
1448 		proto = CEPH_MONC_PROTOCOL;
1449 		break;
1450 	case CEPH_ENTITY_TYPE_OSD:
1451 		proto = CEPH_OSDC_PROTOCOL;
1452 		break;
1453 	case CEPH_ENTITY_TYPE_MDS:
1454 		proto = CEPH_MDSC_PROTOCOL;
1455 		break;
1456 	default:
1457 		BUG();
1458 	}
1459 
1460 	dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
1461 	     con->connect_seq, global_seq, proto);
1462 
1463 	con->out_connect.features =
1464 	    cpu_to_le64(from_msgr(con->msgr)->supported_features);
1465 	con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
1466 	con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
1467 	con->out_connect.global_seq = cpu_to_le32(global_seq);
1468 	con->out_connect.protocol_version = cpu_to_le32(proto);
1469 	con->out_connect.flags = 0;
1470 
1471 	auth_proto = CEPH_AUTH_UNKNOWN;
1472 	auth = get_connect_authorizer(con, &auth_proto);
1473 	if (IS_ERR(auth))
1474 		return PTR_ERR(auth);
1475 
1476 	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1477 	con->out_connect.authorizer_len = auth ?
1478 		cpu_to_le32(auth->authorizer_buf_len) : 0;
1479 
1480 	con_out_kvec_add(con, sizeof (con->out_connect),
1481 					&con->out_connect);
1482 	if (auth && auth->authorizer_buf_len)
1483 		con_out_kvec_add(con, auth->authorizer_buf_len,
1484 					auth->authorizer_buf);
1485 
1486 	con->out_more = 0;
1487 	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1488 
1489 	return 0;
1490 }
1491 
1492 /*
1493  * write as much of pending kvecs to the socket as we can.
1494  *  1 -> done
1495  *  0 -> socket full, but more to do
1496  * <0 -> error
1497  */
write_partial_kvec(struct ceph_connection * con)1498 static int write_partial_kvec(struct ceph_connection *con)
1499 {
1500 	int ret;
1501 
1502 	dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
1503 	while (con->out_kvec_bytes > 0) {
1504 		ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
1505 				       con->out_kvec_left, con->out_kvec_bytes,
1506 				       con->out_more);
1507 		if (ret <= 0)
1508 			goto out;
1509 		con->out_kvec_bytes -= ret;
1510 		if (con->out_kvec_bytes == 0)
1511 			break;            /* done */
1512 
1513 		/* account for full iov entries consumed */
1514 		while (ret >= con->out_kvec_cur->iov_len) {
1515 			BUG_ON(!con->out_kvec_left);
1516 			ret -= con->out_kvec_cur->iov_len;
1517 			con->out_kvec_cur++;
1518 			con->out_kvec_left--;
1519 		}
1520 		/* and for a partially-consumed entry */
1521 		if (ret) {
1522 			con->out_kvec_cur->iov_len -= ret;
1523 			con->out_kvec_cur->iov_base += ret;
1524 		}
1525 	}
1526 	con->out_kvec_left = 0;
1527 	ret = 1;
1528 out:
1529 	dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
1530 	     con->out_kvec_bytes, con->out_kvec_left, ret);
1531 	return ret;  /* done! */
1532 }
1533 
ceph_crc32c_page(u32 crc,struct page * page,unsigned int page_offset,unsigned int length)1534 static u32 ceph_crc32c_page(u32 crc, struct page *page,
1535 				unsigned int page_offset,
1536 				unsigned int length)
1537 {
1538 	char *kaddr;
1539 
1540 	kaddr = kmap(page);
1541 	BUG_ON(kaddr == NULL);
1542 	crc = crc32c(crc, kaddr + page_offset, length);
1543 	kunmap(page);
1544 
1545 	return crc;
1546 }
1547 /*
1548  * Write as much message data payload as we can.  If we finish, queue
1549  * up the footer.
1550  *  1 -> done, footer is now queued in out_kvec[].
1551  *  0 -> socket full, but more to do
1552  * <0 -> error
1553  */
write_partial_message_data(struct ceph_connection * con)1554 static int write_partial_message_data(struct ceph_connection *con)
1555 {
1556 	struct ceph_msg *msg = con->out_msg;
1557 	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1558 	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
1559 	u32 crc;
1560 
1561 	dout("%s %p msg %p\n", __func__, con, msg);
1562 
1563 	if (list_empty(&msg->data))
1564 		return -EINVAL;
1565 
1566 	/*
1567 	 * Iterate through each page that contains data to be
1568 	 * written, and send as much as possible for each.
1569 	 *
1570 	 * If we are calculating the data crc (the default), we will
1571 	 * need to map the page.  If we have no pages, they have
1572 	 * been revoked, so use the zero page.
1573 	 */
1574 	crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0;
1575 	while (cursor->resid) {
1576 		struct page *page;
1577 		size_t page_offset;
1578 		size_t length;
1579 		bool last_piece;
1580 		bool need_crc;
1581 		int ret;
1582 
1583 		page = ceph_msg_data_next(cursor, &page_offset, &length,
1584 					  &last_piece);
1585 		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1586 					length, !last_piece);
1587 		if (ret <= 0) {
1588 			if (do_datacrc)
1589 				msg->footer.data_crc = cpu_to_le32(crc);
1590 
1591 			return ret;
1592 		}
1593 		if (do_datacrc && cursor->need_crc)
1594 			crc = ceph_crc32c_page(crc, page, page_offset, length);
1595 		need_crc = ceph_msg_data_advance(cursor, (size_t)ret);
1596 	}
1597 
1598 	dout("%s %p msg %p done\n", __func__, con, msg);
1599 
1600 	/* prepare and queue up footer, too */
1601 	if (do_datacrc)
1602 		msg->footer.data_crc = cpu_to_le32(crc);
1603 	else
1604 		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1605 	con_out_kvec_reset(con);
1606 	prepare_write_message_footer(con);
1607 
1608 	return 1;	/* must return > 0 to indicate success */
1609 }
1610 
1611 /*
1612  * write some zeros
1613  */
write_partial_skip(struct ceph_connection * con)1614 static int write_partial_skip(struct ceph_connection *con)
1615 {
1616 	int ret;
1617 
1618 	dout("%s %p %d left\n", __func__, con, con->out_skip);
1619 	while (con->out_skip > 0) {
1620 		size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
1621 
1622 		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
1623 		if (ret <= 0)
1624 			goto out;
1625 		con->out_skip -= ret;
1626 	}
1627 	ret = 1;
1628 out:
1629 	return ret;
1630 }
1631 
1632 /*
1633  * Prepare to read connection handshake, or an ack.
1634  */
prepare_read_banner(struct ceph_connection * con)1635 static void prepare_read_banner(struct ceph_connection *con)
1636 {
1637 	dout("prepare_read_banner %p\n", con);
1638 	con->in_base_pos = 0;
1639 }
1640 
prepare_read_connect(struct ceph_connection * con)1641 static void prepare_read_connect(struct ceph_connection *con)
1642 {
1643 	dout("prepare_read_connect %p\n", con);
1644 	con->in_base_pos = 0;
1645 }
1646 
prepare_read_ack(struct ceph_connection * con)1647 static void prepare_read_ack(struct ceph_connection *con)
1648 {
1649 	dout("prepare_read_ack %p\n", con);
1650 	con->in_base_pos = 0;
1651 }
1652 
prepare_read_seq(struct ceph_connection * con)1653 static void prepare_read_seq(struct ceph_connection *con)
1654 {
1655 	dout("prepare_read_seq %p\n", con);
1656 	con->in_base_pos = 0;
1657 	con->in_tag = CEPH_MSGR_TAG_SEQ;
1658 }
1659 
prepare_read_tag(struct ceph_connection * con)1660 static void prepare_read_tag(struct ceph_connection *con)
1661 {
1662 	dout("prepare_read_tag %p\n", con);
1663 	con->in_base_pos = 0;
1664 	con->in_tag = CEPH_MSGR_TAG_READY;
1665 }
1666 
prepare_read_keepalive_ack(struct ceph_connection * con)1667 static void prepare_read_keepalive_ack(struct ceph_connection *con)
1668 {
1669 	dout("prepare_read_keepalive_ack %p\n", con);
1670 	con->in_base_pos = 0;
1671 }
1672 
1673 /*
1674  * Prepare to read a message.
1675  */
prepare_read_message(struct ceph_connection * con)1676 static int prepare_read_message(struct ceph_connection *con)
1677 {
1678 	dout("prepare_read_message %p\n", con);
1679 	BUG_ON(con->in_msg != NULL);
1680 	con->in_base_pos = 0;
1681 	con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
1682 	return 0;
1683 }
1684 
1685 
read_partial(struct ceph_connection * con,int end,int size,void * object)1686 static int read_partial(struct ceph_connection *con,
1687 			int end, int size, void *object)
1688 {
1689 	while (con->in_base_pos < end) {
1690 		int left = end - con->in_base_pos;
1691 		int have = size - left;
1692 		int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
1693 		if (ret <= 0)
1694 			return ret;
1695 		con->in_base_pos += ret;
1696 	}
1697 	return 1;
1698 }
1699 
1700 
1701 /*
1702  * Read all or part of the connect-side handshake on a new connection
1703  */
read_partial_banner(struct ceph_connection * con)1704 static int read_partial_banner(struct ceph_connection *con)
1705 {
1706 	int size;
1707 	int end;
1708 	int ret;
1709 
1710 	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
1711 
1712 	/* peer's banner */
1713 	size = strlen(CEPH_BANNER);
1714 	end = size;
1715 	ret = read_partial(con, end, size, con->in_banner);
1716 	if (ret <= 0)
1717 		goto out;
1718 
1719 	size = sizeof (con->actual_peer_addr);
1720 	end += size;
1721 	ret = read_partial(con, end, size, &con->actual_peer_addr);
1722 	if (ret <= 0)
1723 		goto out;
1724 
1725 	size = sizeof (con->peer_addr_for_me);
1726 	end += size;
1727 	ret = read_partial(con, end, size, &con->peer_addr_for_me);
1728 	if (ret <= 0)
1729 		goto out;
1730 
1731 out:
1732 	return ret;
1733 }
1734 
read_partial_connect(struct ceph_connection * con)1735 static int read_partial_connect(struct ceph_connection *con)
1736 {
1737 	int size;
1738 	int end;
1739 	int ret;
1740 
1741 	dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
1742 
1743 	size = sizeof (con->in_reply);
1744 	end = size;
1745 	ret = read_partial(con, end, size, &con->in_reply);
1746 	if (ret <= 0)
1747 		goto out;
1748 
1749 	size = le32_to_cpu(con->in_reply.authorizer_len);
1750 	end += size;
1751 	ret = read_partial(con, end, size, con->auth_reply_buf);
1752 	if (ret <= 0)
1753 		goto out;
1754 
1755 	dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
1756 	     con, (int)con->in_reply.tag,
1757 	     le32_to_cpu(con->in_reply.connect_seq),
1758 	     le32_to_cpu(con->in_reply.global_seq));
1759 out:
1760 	return ret;
1761 
1762 }
1763 
1764 /*
1765  * Verify the hello banner looks okay.
1766  */
verify_hello(struct ceph_connection * con)1767 static int verify_hello(struct ceph_connection *con)
1768 {
1769 	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1770 		pr_err("connect to %s got bad banner\n",
1771 		       ceph_pr_addr(&con->peer_addr.in_addr));
1772 		con->error_msg = "protocol error, bad banner";
1773 		return -1;
1774 	}
1775 	return 0;
1776 }
1777 
addr_is_blank(struct sockaddr_storage * ss)1778 static bool addr_is_blank(struct sockaddr_storage *ss)
1779 {
1780 	struct in_addr *addr = &((struct sockaddr_in *)ss)->sin_addr;
1781 	struct in6_addr *addr6 = &((struct sockaddr_in6 *)ss)->sin6_addr;
1782 
1783 	switch (ss->ss_family) {
1784 	case AF_INET:
1785 		return addr->s_addr == htonl(INADDR_ANY);
1786 	case AF_INET6:
1787 		return ipv6_addr_any(addr6);
1788 	default:
1789 		return true;
1790 	}
1791 }
1792 
addr_port(struct sockaddr_storage * ss)1793 static int addr_port(struct sockaddr_storage *ss)
1794 {
1795 	switch (ss->ss_family) {
1796 	case AF_INET:
1797 		return ntohs(((struct sockaddr_in *)ss)->sin_port);
1798 	case AF_INET6:
1799 		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
1800 	}
1801 	return 0;
1802 }
1803 
addr_set_port(struct sockaddr_storage * ss,int p)1804 static void addr_set_port(struct sockaddr_storage *ss, int p)
1805 {
1806 	switch (ss->ss_family) {
1807 	case AF_INET:
1808 		((struct sockaddr_in *)ss)->sin_port = htons(p);
1809 		break;
1810 	case AF_INET6:
1811 		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1812 		break;
1813 	}
1814 }
1815 
1816 /*
1817  * Unlike other *_pton function semantics, zero indicates success.
1818  */
ceph_pton(const char * str,size_t len,struct sockaddr_storage * ss,char delim,const char ** ipend)1819 static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
1820 		char delim, const char **ipend)
1821 {
1822 	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
1823 	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
1824 
1825 	memset(ss, 0, sizeof(*ss));
1826 
1827 	if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
1828 		ss->ss_family = AF_INET;
1829 		return 0;
1830 	}
1831 
1832 	if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
1833 		ss->ss_family = AF_INET6;
1834 		return 0;
1835 	}
1836 
1837 	return -EINVAL;
1838 }
1839 
1840 /*
1841  * Extract hostname string and resolve using kernel DNS facility.
1842  */
1843 #ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
ceph_dns_resolve_name(const char * name,size_t namelen,struct sockaddr_storage * ss,char delim,const char ** ipend)1844 static int ceph_dns_resolve_name(const char *name, size_t namelen,
1845 		struct sockaddr_storage *ss, char delim, const char **ipend)
1846 {
1847 	const char *end, *delim_p;
1848 	char *colon_p, *ip_addr = NULL;
1849 	int ip_len, ret;
1850 
1851 	/*
1852 	 * The end of the hostname occurs immediately preceding the delimiter or
1853 	 * the port marker (':') where the delimiter takes precedence.
1854 	 */
1855 	delim_p = memchr(name, delim, namelen);
1856 	colon_p = memchr(name, ':', namelen);
1857 
1858 	if (delim_p && colon_p)
1859 		end = delim_p < colon_p ? delim_p : colon_p;
1860 	else if (!delim_p && colon_p)
1861 		end = colon_p;
1862 	else {
1863 		end = delim_p;
1864 		if (!end) /* case: hostname:/ */
1865 			end = name + namelen;
1866 	}
1867 
1868 	if (end <= name)
1869 		return -EINVAL;
1870 
1871 	/* do dns_resolve upcall */
1872 	ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
1873 	if (ip_len > 0)
1874 		ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
1875 	else
1876 		ret = -ESRCH;
1877 
1878 	kfree(ip_addr);
1879 
1880 	*ipend = end;
1881 
1882 	pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
1883 			ret, ret ? "failed" : ceph_pr_addr(ss));
1884 
1885 	return ret;
1886 }
1887 #else
ceph_dns_resolve_name(const char * name,size_t namelen,struct sockaddr_storage * ss,char delim,const char ** ipend)1888 static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
1889 		struct sockaddr_storage *ss, char delim, const char **ipend)
1890 {
1891 	return -EINVAL;
1892 }
1893 #endif
1894 
1895 /*
1896  * Parse a server name (IP or hostname). If a valid IP address is not found
1897  * then try to extract a hostname to resolve using userspace DNS upcall.
1898  */
ceph_parse_server_name(const char * name,size_t namelen,struct sockaddr_storage * ss,char delim,const char ** ipend)1899 static int ceph_parse_server_name(const char *name, size_t namelen,
1900 			struct sockaddr_storage *ss, char delim, const char **ipend)
1901 {
1902 	int ret;
1903 
1904 	ret = ceph_pton(name, namelen, ss, delim, ipend);
1905 	if (ret)
1906 		ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);
1907 
1908 	return ret;
1909 }
1910 
1911 /*
1912  * Parse an ip[:port] list into an addr array.  Use the default
1913  * monitor port if a port isn't specified.
1914  */
ceph_parse_ips(const char * c,const char * end,struct ceph_entity_addr * addr,int max_count,int * count)1915 int ceph_parse_ips(const char *c, const char *end,
1916 		   struct ceph_entity_addr *addr,
1917 		   int max_count, int *count)
1918 {
1919 	int i, ret = -EINVAL;
1920 	const char *p = c;
1921 
1922 	dout("parse_ips on '%.*s'\n", (int)(end-c), c);
1923 	for (i = 0; i < max_count; i++) {
1924 		const char *ipend;
1925 		struct sockaddr_storage *ss = &addr[i].in_addr;
1926 		int port;
1927 		char delim = ',';
1928 
1929 		if (*p == '[') {
1930 			delim = ']';
1931 			p++;
1932 		}
1933 
1934 		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
1935 		if (ret)
1936 			goto bad;
1937 		ret = -EINVAL;
1938 
1939 		p = ipend;
1940 
1941 		if (delim == ']') {
1942 			if (*p != ']') {
1943 				dout("missing matching ']'\n");
1944 				goto bad;
1945 			}
1946 			p++;
1947 		}
1948 
1949 		/* port? */
1950 		if (p < end && *p == ':') {
1951 			port = 0;
1952 			p++;
1953 			while (p < end && *p >= '0' && *p <= '9') {
1954 				port = (port * 10) + (*p - '0');
1955 				p++;
1956 			}
1957 			if (port == 0)
1958 				port = CEPH_MON_PORT;
1959 			else if (port > 65535)
1960 				goto bad;
1961 		} else {
1962 			port = CEPH_MON_PORT;
1963 		}
1964 
1965 		addr_set_port(ss, port);
1966 
1967 		dout("parse_ips got %s\n", ceph_pr_addr(ss));
1968 
1969 		if (p == end)
1970 			break;
1971 		if (*p != ',')
1972 			goto bad;
1973 		p++;
1974 	}
1975 
1976 	if (p != end)
1977 		goto bad;
1978 
1979 	if (count)
1980 		*count = i + 1;
1981 	return 0;
1982 
1983 bad:
1984 	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
1985 	return ret;
1986 }
1987 EXPORT_SYMBOL(ceph_parse_ips);
1988 
process_banner(struct ceph_connection * con)1989 static int process_banner(struct ceph_connection *con)
1990 {
1991 	dout("process_banner on %p\n", con);
1992 
1993 	if (verify_hello(con) < 0)
1994 		return -1;
1995 
1996 	ceph_decode_addr(&con->actual_peer_addr);
1997 	ceph_decode_addr(&con->peer_addr_for_me);
1998 
1999 	/*
2000 	 * Make sure the other end is who we wanted.  note that the other
2001 	 * end may not yet know their ip address, so if it's 0.0.0.0, give
2002 	 * them the benefit of the doubt.
2003 	 */
2004 	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
2005 		   sizeof(con->peer_addr)) != 0 &&
2006 	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
2007 	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
2008 		pr_warn("wrong peer, want %s/%d, got %s/%d\n",
2009 			ceph_pr_addr(&con->peer_addr.in_addr),
2010 			(int)le32_to_cpu(con->peer_addr.nonce),
2011 			ceph_pr_addr(&con->actual_peer_addr.in_addr),
2012 			(int)le32_to_cpu(con->actual_peer_addr.nonce));
2013 		con->error_msg = "wrong peer at address";
2014 		return -1;
2015 	}
2016 
2017 	/*
2018 	 * did we learn our address?
2019 	 */
2020 	if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
2021 		int port = addr_port(&con->msgr->inst.addr.in_addr);
2022 
2023 		memcpy(&con->msgr->inst.addr.in_addr,
2024 		       &con->peer_addr_for_me.in_addr,
2025 		       sizeof(con->peer_addr_for_me.in_addr));
2026 		addr_set_port(&con->msgr->inst.addr.in_addr, port);
2027 		encode_my_addr(con->msgr);
2028 		dout("process_banner learned my addr is %s\n",
2029 		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
2030 	}
2031 
2032 	return 0;
2033 }
2034 
process_connect(struct ceph_connection * con)2035 static int process_connect(struct ceph_connection *con)
2036 {
2037 	u64 sup_feat = from_msgr(con->msgr)->supported_features;
2038 	u64 req_feat = from_msgr(con->msgr)->required_features;
2039 	u64 server_feat = ceph_sanitize_features(
2040 				le64_to_cpu(con->in_reply.features));
2041 	int ret;
2042 
2043 	dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
2044 
2045 	switch (con->in_reply.tag) {
2046 	case CEPH_MSGR_TAG_FEATURES:
2047 		pr_err("%s%lld %s feature set mismatch,"
2048 		       " my %llx < server's %llx, missing %llx\n",
2049 		       ENTITY_NAME(con->peer_name),
2050 		       ceph_pr_addr(&con->peer_addr.in_addr),
2051 		       sup_feat, server_feat, server_feat & ~sup_feat);
2052 		con->error_msg = "missing required protocol features";
2053 		reset_connection(con);
2054 		return -1;
2055 
2056 	case CEPH_MSGR_TAG_BADPROTOVER:
2057 		pr_err("%s%lld %s protocol version mismatch,"
2058 		       " my %d != server's %d\n",
2059 		       ENTITY_NAME(con->peer_name),
2060 		       ceph_pr_addr(&con->peer_addr.in_addr),
2061 		       le32_to_cpu(con->out_connect.protocol_version),
2062 		       le32_to_cpu(con->in_reply.protocol_version));
2063 		con->error_msg = "protocol version mismatch";
2064 		reset_connection(con);
2065 		return -1;
2066 
2067 	case CEPH_MSGR_TAG_BADAUTHORIZER:
2068 		con->auth_retry++;
2069 		dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
2070 		     con->auth_retry);
2071 		if (con->auth_retry == 2) {
2072 			con->error_msg = "connect authorization failure";
2073 			return -1;
2074 		}
2075 		con_out_kvec_reset(con);
2076 		ret = prepare_write_connect(con);
2077 		if (ret < 0)
2078 			return ret;
2079 		prepare_read_connect(con);
2080 		break;
2081 
2082 	case CEPH_MSGR_TAG_RESETSESSION:
2083 		/*
2084 		 * If we connected with a large connect_seq but the peer
2085 		 * has no record of a session with us (no connection, or
2086 		 * connect_seq == 0), they will send RESETSESION to indicate
2087 		 * that they must have reset their session, and may have
2088 		 * dropped messages.
2089 		 */
2090 		dout("process_connect got RESET peer seq %u\n",
2091 		     le32_to_cpu(con->in_reply.connect_seq));
2092 		pr_err("%s%lld %s connection reset\n",
2093 		       ENTITY_NAME(con->peer_name),
2094 		       ceph_pr_addr(&con->peer_addr.in_addr));
2095 		reset_connection(con);
2096 		con_out_kvec_reset(con);
2097 		ret = prepare_write_connect(con);
2098 		if (ret < 0)
2099 			return ret;
2100 		prepare_read_connect(con);
2101 
2102 		/* Tell ceph about it. */
2103 		mutex_unlock(&con->mutex);
2104 		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
2105 		if (con->ops->peer_reset)
2106 			con->ops->peer_reset(con);
2107 		mutex_lock(&con->mutex);
2108 		if (con->state != CON_STATE_NEGOTIATING)
2109 			return -EAGAIN;
2110 		break;
2111 
2112 	case CEPH_MSGR_TAG_RETRY_SESSION:
2113 		/*
2114 		 * If we sent a smaller connect_seq than the peer has, try
2115 		 * again with a larger value.
2116 		 */
2117 		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
2118 		     le32_to_cpu(con->out_connect.connect_seq),
2119 		     le32_to_cpu(con->in_reply.connect_seq));
2120 		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2121 		con_out_kvec_reset(con);
2122 		ret = prepare_write_connect(con);
2123 		if (ret < 0)
2124 			return ret;
2125 		prepare_read_connect(con);
2126 		break;
2127 
2128 	case CEPH_MSGR_TAG_RETRY_GLOBAL:
2129 		/*
2130 		 * If we sent a smaller global_seq than the peer has, try
2131 		 * again with a larger value.
2132 		 */
2133 		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
2134 		     con->peer_global_seq,
2135 		     le32_to_cpu(con->in_reply.global_seq));
2136 		get_global_seq(con->msgr,
2137 			       le32_to_cpu(con->in_reply.global_seq));
2138 		con_out_kvec_reset(con);
2139 		ret = prepare_write_connect(con);
2140 		if (ret < 0)
2141 			return ret;
2142 		prepare_read_connect(con);
2143 		break;
2144 
2145 	case CEPH_MSGR_TAG_SEQ:
2146 	case CEPH_MSGR_TAG_READY:
2147 		if (req_feat & ~server_feat) {
2148 			pr_err("%s%lld %s protocol feature mismatch,"
2149 			       " my required %llx > server's %llx, need %llx\n",
2150 			       ENTITY_NAME(con->peer_name),
2151 			       ceph_pr_addr(&con->peer_addr.in_addr),
2152 			       req_feat, server_feat, req_feat & ~server_feat);
2153 			con->error_msg = "missing required protocol features";
2154 			reset_connection(con);
2155 			return -1;
2156 		}
2157 
2158 		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2159 		con->state = CON_STATE_OPEN;
2160 		con->auth_retry = 0;    /* we authenticated; clear flag */
2161 		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
2162 		con->connect_seq++;
2163 		con->peer_features = server_feat;
2164 		dout("process_connect got READY gseq %d cseq %d (%d)\n",
2165 		     con->peer_global_seq,
2166 		     le32_to_cpu(con->in_reply.connect_seq),
2167 		     con->connect_seq);
2168 		WARN_ON(con->connect_seq !=
2169 			le32_to_cpu(con->in_reply.connect_seq));
2170 
2171 		if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
2172 			con_flag_set(con, CON_FLAG_LOSSYTX);
2173 
2174 		con->delay = 0;      /* reset backoff memory */
2175 
2176 		if (con->in_reply.tag == CEPH_MSGR_TAG_SEQ) {
2177 			prepare_write_seq(con);
2178 			prepare_read_seq(con);
2179 		} else {
2180 			prepare_read_tag(con);
2181 		}
2182 		break;
2183 
2184 	case CEPH_MSGR_TAG_WAIT:
2185 		/*
2186 		 * If there is a connection race (we are opening
2187 		 * connections to each other), one of us may just have
2188 		 * to WAIT.  This shouldn't happen if we are the
2189 		 * client.
2190 		 */
2191 		con->error_msg = "protocol error, got WAIT as client";
2192 		return -1;
2193 
2194 	default:
2195 		con->error_msg = "protocol error, garbage tag during connect";
2196 		return -1;
2197 	}
2198 	return 0;
2199 }
2200 
2201 
2202 /*
2203  * read (part of) an ack
2204  */
read_partial_ack(struct ceph_connection * con)2205 static int read_partial_ack(struct ceph_connection *con)
2206 {
2207 	int size = sizeof (con->in_temp_ack);
2208 	int end = size;
2209 
2210 	return read_partial(con, end, size, &con->in_temp_ack);
2211 }
2212 
2213 /*
2214  * We can finally discard anything that's been acked.
2215  */
process_ack(struct ceph_connection * con)2216 static void process_ack(struct ceph_connection *con)
2217 {
2218 	struct ceph_msg *m;
2219 	u64 ack = le64_to_cpu(con->in_temp_ack);
2220 	u64 seq;
2221 
2222 	while (!list_empty(&con->out_sent)) {
2223 		m = list_first_entry(&con->out_sent, struct ceph_msg,
2224 				     list_head);
2225 		seq = le64_to_cpu(m->hdr.seq);
2226 		if (seq > ack)
2227 			break;
2228 		dout("got ack for seq %llu type %d at %p\n", seq,
2229 		     le16_to_cpu(m->hdr.type), m);
2230 		m->ack_stamp = jiffies;
2231 		ceph_msg_remove(m);
2232 	}
2233 	prepare_read_tag(con);
2234 }
2235 
2236 
read_partial_message_section(struct ceph_connection * con,struct kvec * section,unsigned int sec_len,u32 * crc)2237 static int read_partial_message_section(struct ceph_connection *con,
2238 					struct kvec *section,
2239 					unsigned int sec_len, u32 *crc)
2240 {
2241 	int ret, left;
2242 
2243 	BUG_ON(!section);
2244 
2245 	while (section->iov_len < sec_len) {
2246 		BUG_ON(section->iov_base == NULL);
2247 		left = sec_len - section->iov_len;
2248 		ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
2249 				       section->iov_len, left);
2250 		if (ret <= 0)
2251 			return ret;
2252 		section->iov_len += ret;
2253 	}
2254 	if (section->iov_len == sec_len)
2255 		*crc = crc32c(0, section->iov_base, section->iov_len);
2256 
2257 	return 1;
2258 }
2259 
read_partial_msg_data(struct ceph_connection * con)2260 static int read_partial_msg_data(struct ceph_connection *con)
2261 {
2262 	struct ceph_msg *msg = con->in_msg;
2263 	struct ceph_msg_data_cursor *cursor = &msg->cursor;
2264 	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
2265 	struct page *page;
2266 	size_t page_offset;
2267 	size_t length;
2268 	u32 crc = 0;
2269 	int ret;
2270 
2271 	BUG_ON(!msg);
2272 	if (list_empty(&msg->data))
2273 		return -EIO;
2274 
2275 	if (do_datacrc)
2276 		crc = con->in_data_crc;
2277 	while (cursor->resid) {
2278 		page = ceph_msg_data_next(cursor, &page_offset, &length, NULL);
2279 		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
2280 		if (ret <= 0) {
2281 			if (do_datacrc)
2282 				con->in_data_crc = crc;
2283 
2284 			return ret;
2285 		}
2286 
2287 		if (do_datacrc)
2288 			crc = ceph_crc32c_page(crc, page, page_offset, ret);
2289 		(void) ceph_msg_data_advance(cursor, (size_t)ret);
2290 	}
2291 	if (do_datacrc)
2292 		con->in_data_crc = crc;
2293 
2294 	return 1;	/* must return > 0 to indicate success */
2295 }
2296 
2297 /*
2298  * read (part of) a message.
2299  */
2300 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);
2301 
read_partial_message(struct ceph_connection * con)2302 static int read_partial_message(struct ceph_connection *con)
2303 {
2304 	struct ceph_msg *m = con->in_msg;
2305 	int size;
2306 	int end;
2307 	int ret;
2308 	unsigned int front_len, middle_len, data_len;
2309 	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
2310 	bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
2311 	u64 seq;
2312 	u32 crc;
2313 
2314 	dout("read_partial_message con %p msg %p\n", con, m);
2315 
2316 	/* header */
2317 	size = sizeof (con->in_hdr);
2318 	end = size;
2319 	ret = read_partial(con, end, size, &con->in_hdr);
2320 	if (ret <= 0)
2321 		return ret;
2322 
2323 	crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
2324 	if (cpu_to_le32(crc) != con->in_hdr.crc) {
2325 		pr_err("read_partial_message bad hdr crc %u != expected %u\n",
2326 		       crc, con->in_hdr.crc);
2327 		return -EBADMSG;
2328 	}
2329 
2330 	front_len = le32_to_cpu(con->in_hdr.front_len);
2331 	if (front_len > CEPH_MSG_MAX_FRONT_LEN)
2332 		return -EIO;
2333 	middle_len = le32_to_cpu(con->in_hdr.middle_len);
2334 	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
2335 		return -EIO;
2336 	data_len = le32_to_cpu(con->in_hdr.data_len);
2337 	if (data_len > CEPH_MSG_MAX_DATA_LEN)
2338 		return -EIO;
2339 
2340 	/* verify seq# */
2341 	seq = le64_to_cpu(con->in_hdr.seq);
2342 	if ((s64)seq - (s64)con->in_seq < 1) {
2343 		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2344 			ENTITY_NAME(con->peer_name),
2345 			ceph_pr_addr(&con->peer_addr.in_addr),
2346 			seq, con->in_seq + 1);
2347 		con->in_base_pos = -front_len - middle_len - data_len -
2348 			sizeof_footer(con);
2349 		con->in_tag = CEPH_MSGR_TAG_READY;
2350 		return 1;
2351 	} else if ((s64)seq - (s64)con->in_seq > 1) {
2352 		pr_err("read_partial_message bad seq %lld expected %lld\n",
2353 		       seq, con->in_seq + 1);
2354 		con->error_msg = "bad message sequence # for incoming message";
2355 		return -EBADE;
2356 	}
2357 
2358 	/* allocate message? */
2359 	if (!con->in_msg) {
2360 		int skip = 0;
2361 
2362 		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2363 		     front_len, data_len);
2364 		ret = ceph_con_in_msg_alloc(con, &skip);
2365 		if (ret < 0)
2366 			return ret;
2367 
2368 		BUG_ON(!con->in_msg ^ skip);
2369 		if (skip) {
2370 			/* skip this message */
2371 			dout("alloc_msg said skip message\n");
2372 			con->in_base_pos = -front_len - middle_len - data_len -
2373 				sizeof_footer(con);
2374 			con->in_tag = CEPH_MSGR_TAG_READY;
2375 			con->in_seq++;
2376 			return 1;
2377 		}
2378 
2379 		BUG_ON(!con->in_msg);
2380 		BUG_ON(con->in_msg->con != con);
2381 		m = con->in_msg;
2382 		m->front.iov_len = 0;    /* haven't read it yet */
2383 		if (m->middle)
2384 			m->middle->vec.iov_len = 0;
2385 
2386 		/* prepare for data payload, if any */
2387 
2388 		if (data_len)
2389 			prepare_message_data(con->in_msg, data_len);
2390 	}
2391 
2392 	/* front */
2393 	ret = read_partial_message_section(con, &m->front, front_len,
2394 					   &con->in_front_crc);
2395 	if (ret <= 0)
2396 		return ret;
2397 
2398 	/* middle */
2399 	if (m->middle) {
2400 		ret = read_partial_message_section(con, &m->middle->vec,
2401 						   middle_len,
2402 						   &con->in_middle_crc);
2403 		if (ret <= 0)
2404 			return ret;
2405 	}
2406 
2407 	/* (page) data */
2408 	if (data_len) {
2409 		ret = read_partial_msg_data(con);
2410 		if (ret <= 0)
2411 			return ret;
2412 	}
2413 
2414 	/* footer */
2415 	if (need_sign)
2416 		size = sizeof(m->footer);
2417 	else
2418 		size = sizeof(m->old_footer);
2419 
2420 	end += size;
2421 	ret = read_partial(con, end, size, &m->footer);
2422 	if (ret <= 0)
2423 		return ret;
2424 
2425 	if (!need_sign) {
2426 		m->footer.flags = m->old_footer.flags;
2427 		m->footer.sig = 0;
2428 	}
2429 
2430 	dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
2431 	     m, front_len, m->footer.front_crc, middle_len,
2432 	     m->footer.middle_crc, data_len, m->footer.data_crc);
2433 
2434 	/* crc ok? */
2435 	if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
2436 		pr_err("read_partial_message %p front crc %u != exp. %u\n",
2437 		       m, con->in_front_crc, m->footer.front_crc);
2438 		return -EBADMSG;
2439 	}
2440 	if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
2441 		pr_err("read_partial_message %p middle crc %u != exp %u\n",
2442 		       m, con->in_middle_crc, m->footer.middle_crc);
2443 		return -EBADMSG;
2444 	}
2445 	if (do_datacrc &&
2446 	    (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
2447 	    con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
2448 		pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
2449 		       con->in_data_crc, le32_to_cpu(m->footer.data_crc));
2450 		return -EBADMSG;
2451 	}
2452 
2453 	if (need_sign && con->ops->check_message_signature &&
2454 	    con->ops->check_message_signature(m)) {
2455 		pr_err("read_partial_message %p signature check failed\n", m);
2456 		return -EBADMSG;
2457 	}
2458 
2459 	return 1; /* done! */
2460 }
2461 
2462 /*
2463  * Process message.  This happens in the worker thread.  The callback should
2464  * be careful not to do anything that waits on other incoming messages or it
2465  * may deadlock.
2466  */
process_message(struct ceph_connection * con)2467 static void process_message(struct ceph_connection *con)
2468 {
2469 	struct ceph_msg *msg = con->in_msg;
2470 
2471 	BUG_ON(con->in_msg->con != con);
2472 	con->in_msg = NULL;
2473 
2474 	/* if first message, set peer_name */
2475 	if (con->peer_name.type == 0)
2476 		con->peer_name = msg->hdr.src;
2477 
2478 	con->in_seq++;
2479 	mutex_unlock(&con->mutex);
2480 
2481 	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
2482 	     msg, le64_to_cpu(msg->hdr.seq),
2483 	     ENTITY_NAME(msg->hdr.src),
2484 	     le16_to_cpu(msg->hdr.type),
2485 	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2486 	     le32_to_cpu(msg->hdr.front_len),
2487 	     le32_to_cpu(msg->hdr.data_len),
2488 	     con->in_front_crc, con->in_middle_crc, con->in_data_crc);
2489 	con->ops->dispatch(con, msg);
2490 
2491 	mutex_lock(&con->mutex);
2492 }
2493 
read_keepalive_ack(struct ceph_connection * con)2494 static int read_keepalive_ack(struct ceph_connection *con)
2495 {
2496 	struct ceph_timespec ceph_ts;
2497 	size_t size = sizeof(ceph_ts);
2498 	int ret = read_partial(con, size, size, &ceph_ts);
2499 	if (ret <= 0)
2500 		return ret;
2501 	ceph_decode_timespec(&con->last_keepalive_ack, &ceph_ts);
2502 	prepare_read_tag(con);
2503 	return 1;
2504 }
2505 
2506 /*
2507  * Write something to the socket.  Called in a worker thread when the
2508  * socket appears to be writeable and we have something ready to send.
2509  */
try_write(struct ceph_connection * con)2510 static int try_write(struct ceph_connection *con)
2511 {
2512 	int ret = 1;
2513 
2514 	dout("try_write start %p state %lu\n", con, con->state);
2515 
2516 more:
2517 	dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
2518 
2519 	/* open the socket first? */
2520 	if (con->state == CON_STATE_PREOPEN) {
2521 		BUG_ON(con->sock);
2522 		con->state = CON_STATE_CONNECTING;
2523 
2524 		con_out_kvec_reset(con);
2525 		prepare_write_banner(con);
2526 		prepare_read_banner(con);
2527 
2528 		BUG_ON(con->in_msg);
2529 		con->in_tag = CEPH_MSGR_TAG_READY;
2530 		dout("try_write initiating connect on %p new state %lu\n",
2531 		     con, con->state);
2532 		ret = ceph_tcp_connect(con);
2533 		if (ret < 0) {
2534 			con->error_msg = "connect error";
2535 			goto out;
2536 		}
2537 	}
2538 
2539 more_kvec:
2540 	/* kvec data queued? */
2541 	if (con->out_kvec_left) {
2542 		ret = write_partial_kvec(con);
2543 		if (ret <= 0)
2544 			goto out;
2545 	}
2546 	if (con->out_skip) {
2547 		ret = write_partial_skip(con);
2548 		if (ret <= 0)
2549 			goto out;
2550 	}
2551 
2552 	/* msg pages? */
2553 	if (con->out_msg) {
2554 		if (con->out_msg_done) {
2555 			ceph_msg_put(con->out_msg);
2556 			con->out_msg = NULL;   /* we're done with this one */
2557 			goto do_next;
2558 		}
2559 
2560 		ret = write_partial_message_data(con);
2561 		if (ret == 1)
2562 			goto more_kvec;  /* we need to send the footer, too! */
2563 		if (ret == 0)
2564 			goto out;
2565 		if (ret < 0) {
2566 			dout("try_write write_partial_message_data err %d\n",
2567 			     ret);
2568 			goto out;
2569 		}
2570 	}
2571 
2572 do_next:
2573 	if (con->state == CON_STATE_OPEN) {
2574 		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
2575 			prepare_write_keepalive(con);
2576 			goto more;
2577 		}
2578 		/* is anything else pending? */
2579 		if (!list_empty(&con->out_queue)) {
2580 			prepare_write_message(con);
2581 			goto more;
2582 		}
2583 		if (con->in_seq > con->in_seq_acked) {
2584 			prepare_write_ack(con);
2585 			goto more;
2586 		}
2587 	}
2588 
2589 	/* Nothing to do! */
2590 	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2591 	dout("try_write nothing else to write.\n");
2592 	ret = 0;
2593 out:
2594 	dout("try_write done on %p ret %d\n", con, ret);
2595 	return ret;
2596 }
2597 
2598 
2599 
2600 /*
2601  * Read what we can from the socket.
2602  */
try_read(struct ceph_connection * con)2603 static int try_read(struct ceph_connection *con)
2604 {
2605 	int ret = -1;
2606 
2607 more:
2608 	dout("try_read start on %p state %lu\n", con, con->state);
2609 	if (con->state != CON_STATE_CONNECTING &&
2610 	    con->state != CON_STATE_NEGOTIATING &&
2611 	    con->state != CON_STATE_OPEN)
2612 		return 0;
2613 
2614 	BUG_ON(!con->sock);
2615 
2616 	dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
2617 	     con->in_base_pos);
2618 
2619 	if (con->state == CON_STATE_CONNECTING) {
2620 		dout("try_read connecting\n");
2621 		ret = read_partial_banner(con);
2622 		if (ret <= 0)
2623 			goto out;
2624 		ret = process_banner(con);
2625 		if (ret < 0)
2626 			goto out;
2627 
2628 		con->state = CON_STATE_NEGOTIATING;
2629 
2630 		/*
2631 		 * Received banner is good, exchange connection info.
2632 		 * Do not reset out_kvec, as sending our banner raced
2633 		 * with receiving peer banner after connect completed.
2634 		 */
2635 		ret = prepare_write_connect(con);
2636 		if (ret < 0)
2637 			goto out;
2638 		prepare_read_connect(con);
2639 
2640 		/* Send connection info before awaiting response */
2641 		goto out;
2642 	}
2643 
2644 	if (con->state == CON_STATE_NEGOTIATING) {
2645 		dout("try_read negotiating\n");
2646 		ret = read_partial_connect(con);
2647 		if (ret <= 0)
2648 			goto out;
2649 		ret = process_connect(con);
2650 		if (ret < 0)
2651 			goto out;
2652 		goto more;
2653 	}
2654 
2655 	WARN_ON(con->state != CON_STATE_OPEN);
2656 
2657 	if (con->in_base_pos < 0) {
2658 		/*
2659 		 * skipping + discarding content.
2660 		 *
2661 		 * FIXME: there must be a better way to do this!
2662 		 */
2663 		static char buf[SKIP_BUF_SIZE];
2664 		int skip = min((int) sizeof (buf), -con->in_base_pos);
2665 
2666 		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
2667 		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
2668 		if (ret <= 0)
2669 			goto out;
2670 		con->in_base_pos += ret;
2671 		if (con->in_base_pos)
2672 			goto more;
2673 	}
2674 	if (con->in_tag == CEPH_MSGR_TAG_READY) {
2675 		/*
2676 		 * what's next?
2677 		 */
2678 		ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
2679 		if (ret <= 0)
2680 			goto out;
2681 		dout("try_read got tag %d\n", (int)con->in_tag);
2682 		switch (con->in_tag) {
2683 		case CEPH_MSGR_TAG_MSG:
2684 			prepare_read_message(con);
2685 			break;
2686 		case CEPH_MSGR_TAG_ACK:
2687 			prepare_read_ack(con);
2688 			break;
2689 		case CEPH_MSGR_TAG_KEEPALIVE2_ACK:
2690 			prepare_read_keepalive_ack(con);
2691 			break;
2692 		case CEPH_MSGR_TAG_CLOSE:
2693 			con_close_socket(con);
2694 			con->state = CON_STATE_CLOSED;
2695 			goto out;
2696 		default:
2697 			goto bad_tag;
2698 		}
2699 	}
2700 	if (con->in_tag == CEPH_MSGR_TAG_MSG) {
2701 		ret = read_partial_message(con);
2702 		if (ret <= 0) {
2703 			switch (ret) {
2704 			case -EBADMSG:
2705 				con->error_msg = "bad crc/signature";
2706 				/* fall through */
2707 			case -EBADE:
2708 				ret = -EIO;
2709 				break;
2710 			case -EIO:
2711 				con->error_msg = "io error";
2712 				break;
2713 			}
2714 			goto out;
2715 		}
2716 		if (con->in_tag == CEPH_MSGR_TAG_READY)
2717 			goto more;
2718 		process_message(con);
2719 		if (con->state == CON_STATE_OPEN)
2720 			prepare_read_tag(con);
2721 		goto more;
2722 	}
2723 	if (con->in_tag == CEPH_MSGR_TAG_ACK ||
2724 	    con->in_tag == CEPH_MSGR_TAG_SEQ) {
2725 		/*
2726 		 * the final handshake seq exchange is semantically
2727 		 * equivalent to an ACK
2728 		 */
2729 		ret = read_partial_ack(con);
2730 		if (ret <= 0)
2731 			goto out;
2732 		process_ack(con);
2733 		goto more;
2734 	}
2735 	if (con->in_tag == CEPH_MSGR_TAG_KEEPALIVE2_ACK) {
2736 		ret = read_keepalive_ack(con);
2737 		if (ret <= 0)
2738 			goto out;
2739 		goto more;
2740 	}
2741 
2742 out:
2743 	dout("try_read done on %p ret %d\n", con, ret);
2744 	return ret;
2745 
2746 bad_tag:
2747 	pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
2748 	con->error_msg = "protocol error, garbage tag";
2749 	ret = -1;
2750 	goto out;
2751 }
2752 
2753 
2754 /*
2755  * Atomically queue work on a connection after the specified delay.
2756  * Bump @con reference to avoid races with connection teardown.
2757  * Returns 0 if work was queued, or an error code otherwise.
2758  */
queue_con_delay(struct ceph_connection * con,unsigned long delay)2759 static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
2760 {
2761 	if (!con->ops->get(con)) {
2762 		dout("%s %p ref count 0\n", __func__, con);
2763 		return -ENOENT;
2764 	}
2765 
2766 	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
2767 		dout("%s %p - already queued\n", __func__, con);
2768 		con->ops->put(con);
2769 		return -EBUSY;
2770 	}
2771 
2772 	dout("%s %p %lu\n", __func__, con, delay);
2773 	return 0;
2774 }
2775 
queue_con(struct ceph_connection * con)2776 static void queue_con(struct ceph_connection *con)
2777 {
2778 	(void) queue_con_delay(con, 0);
2779 }
2780 
cancel_con(struct ceph_connection * con)2781 static void cancel_con(struct ceph_connection *con)
2782 {
2783 	if (cancel_delayed_work(&con->work)) {
2784 		dout("%s %p\n", __func__, con);
2785 		con->ops->put(con);
2786 	}
2787 }
2788 
con_sock_closed(struct ceph_connection * con)2789 static bool con_sock_closed(struct ceph_connection *con)
2790 {
2791 	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2792 		return false;
2793 
2794 #define CASE(x)								\
2795 	case CON_STATE_ ## x:						\
2796 		con->error_msg = "socket closed (con state " #x ")";	\
2797 		break;
2798 
2799 	switch (con->state) {
2800 	CASE(CLOSED);
2801 	CASE(PREOPEN);
2802 	CASE(CONNECTING);
2803 	CASE(NEGOTIATING);
2804 	CASE(OPEN);
2805 	CASE(STANDBY);
2806 	default:
2807 		pr_warn("%s con %p unrecognized state %lu\n",
2808 			__func__, con, con->state);
2809 		con->error_msg = "unrecognized con state";
2810 		BUG();
2811 		break;
2812 	}
2813 #undef CASE
2814 
2815 	return true;
2816 }
2817 
con_backoff(struct ceph_connection * con)2818 static bool con_backoff(struct ceph_connection *con)
2819 {
2820 	int ret;
2821 
2822 	if (!con_flag_test_and_clear(con, CON_FLAG_BACKOFF))
2823 		return false;
2824 
2825 	ret = queue_con_delay(con, round_jiffies_relative(con->delay));
2826 	if (ret) {
2827 		dout("%s: con %p FAILED to back off %lu\n", __func__,
2828 			con, con->delay);
2829 		BUG_ON(ret == -ENOENT);
2830 		con_flag_set(con, CON_FLAG_BACKOFF);
2831 	}
2832 
2833 	return true;
2834 }
2835 
2836 /* Finish fault handling; con->mutex must *not* be held here */
2837 
con_fault_finish(struct ceph_connection * con)2838 static void con_fault_finish(struct ceph_connection *con)
2839 {
2840 	/*
2841 	 * in case we faulted due to authentication, invalidate our
2842 	 * current tickets so that we can get new ones.
2843 	 */
2844 	if (con->auth_retry && con->ops->invalidate_authorizer) {
2845 		dout("calling invalidate_authorizer()\n");
2846 		con->ops->invalidate_authorizer(con);
2847 	}
2848 
2849 	if (con->ops->fault)
2850 		con->ops->fault(con);
2851 }
2852 
2853 /*
2854  * Do some work on a connection.  Drop a connection ref when we're done.
2855  */
ceph_con_workfn(struct work_struct * work)2856 static void ceph_con_workfn(struct work_struct *work)
2857 {
2858 	struct ceph_connection *con = container_of(work, struct ceph_connection,
2859 						   work.work);
2860 	bool fault;
2861 
2862 	mutex_lock(&con->mutex);
2863 	while (true) {
2864 		int ret;
2865 
2866 		if ((fault = con_sock_closed(con))) {
2867 			dout("%s: con %p SOCK_CLOSED\n", __func__, con);
2868 			break;
2869 		}
2870 		if (con_backoff(con)) {
2871 			dout("%s: con %p BACKOFF\n", __func__, con);
2872 			break;
2873 		}
2874 		if (con->state == CON_STATE_STANDBY) {
2875 			dout("%s: con %p STANDBY\n", __func__, con);
2876 			break;
2877 		}
2878 		if (con->state == CON_STATE_CLOSED) {
2879 			dout("%s: con %p CLOSED\n", __func__, con);
2880 			BUG_ON(con->sock);
2881 			break;
2882 		}
2883 		if (con->state == CON_STATE_PREOPEN) {
2884 			dout("%s: con %p PREOPEN\n", __func__, con);
2885 			BUG_ON(con->sock);
2886 		}
2887 
2888 		ret = try_read(con);
2889 		if (ret < 0) {
2890 			if (ret == -EAGAIN)
2891 				continue;
2892 			if (!con->error_msg)
2893 				con->error_msg = "socket error on read";
2894 			fault = true;
2895 			break;
2896 		}
2897 
2898 		ret = try_write(con);
2899 		if (ret < 0) {
2900 			if (ret == -EAGAIN)
2901 				continue;
2902 			if (!con->error_msg)
2903 				con->error_msg = "socket error on write";
2904 			fault = true;
2905 		}
2906 
2907 		break;	/* If we make it to here, we're done */
2908 	}
2909 	if (fault)
2910 		con_fault(con);
2911 	mutex_unlock(&con->mutex);
2912 
2913 	if (fault)
2914 		con_fault_finish(con);
2915 
2916 	con->ops->put(con);
2917 }
2918 
2919 /*
2920  * Generic error/fault handler.  A retry mechanism is used with
2921  * exponential backoff
2922  */
con_fault(struct ceph_connection * con)2923 static void con_fault(struct ceph_connection *con)
2924 {
2925 	dout("fault %p state %lu to peer %s\n",
2926 	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
2927 
2928 	pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2929 		ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
2930 	con->error_msg = NULL;
2931 
2932 	WARN_ON(con->state != CON_STATE_CONNECTING &&
2933 	       con->state != CON_STATE_NEGOTIATING &&
2934 	       con->state != CON_STATE_OPEN);
2935 
2936 	con_close_socket(con);
2937 
2938 	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2939 		dout("fault on LOSSYTX channel, marking CLOSED\n");
2940 		con->state = CON_STATE_CLOSED;
2941 		return;
2942 	}
2943 
2944 	if (con->in_msg) {
2945 		BUG_ON(con->in_msg->con != con);
2946 		ceph_msg_put(con->in_msg);
2947 		con->in_msg = NULL;
2948 	}
2949 
2950 	/* Requeue anything that hasn't been acked */
2951 	list_splice_init(&con->out_sent, &con->out_queue);
2952 
2953 	/* If there are no messages queued or keepalive pending, place
2954 	 * the connection in a STANDBY state */
2955 	if (list_empty(&con->out_queue) &&
2956 	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
2957 		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2958 		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2959 		con->state = CON_STATE_STANDBY;
2960 	} else {
2961 		/* retry after a delay. */
2962 		con->state = CON_STATE_PREOPEN;
2963 		if (con->delay == 0)
2964 			con->delay = BASE_DELAY_INTERVAL;
2965 		else if (con->delay < MAX_DELAY_INTERVAL)
2966 			con->delay *= 2;
2967 		con_flag_set(con, CON_FLAG_BACKOFF);
2968 		queue_con(con);
2969 	}
2970 }
2971 
2972 
2973 
2974 /*
2975  * initialize a new messenger instance
2976  */
ceph_messenger_init(struct ceph_messenger * msgr,struct ceph_entity_addr * myaddr)2977 void ceph_messenger_init(struct ceph_messenger *msgr,
2978 			 struct ceph_entity_addr *myaddr)
2979 {
2980 	spin_lock_init(&msgr->global_seq_lock);
2981 
2982 	if (myaddr)
2983 		msgr->inst.addr = *myaddr;
2984 
2985 	/* select a random nonce */
2986 	msgr->inst.addr.type = 0;
2987 	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2988 	encode_my_addr(msgr);
2989 
2990 	atomic_set(&msgr->stopping, 0);
2991 	write_pnet(&msgr->net, get_net(current->nsproxy->net_ns));
2992 
2993 	dout("%s %p\n", __func__, msgr);
2994 }
2995 EXPORT_SYMBOL(ceph_messenger_init);
2996 
ceph_messenger_fini(struct ceph_messenger * msgr)2997 void ceph_messenger_fini(struct ceph_messenger *msgr)
2998 {
2999 	put_net(read_pnet(&msgr->net));
3000 }
3001 EXPORT_SYMBOL(ceph_messenger_fini);
3002 
msg_con_set(struct ceph_msg * msg,struct ceph_connection * con)3003 static void msg_con_set(struct ceph_msg *msg, struct ceph_connection *con)
3004 {
3005 	if (msg->con)
3006 		msg->con->ops->put(msg->con);
3007 
3008 	msg->con = con ? con->ops->get(con) : NULL;
3009 	BUG_ON(msg->con != con);
3010 }
3011 
clear_standby(struct ceph_connection * con)3012 static void clear_standby(struct ceph_connection *con)
3013 {
3014 	/* come back from STANDBY? */
3015 	if (con->state == CON_STATE_STANDBY) {
3016 		dout("clear_standby %p and ++connect_seq\n", con);
3017 		con->state = CON_STATE_PREOPEN;
3018 		con->connect_seq++;
3019 		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
3020 		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
3021 	}
3022 }
3023 
3024 /*
3025  * Queue up an outgoing message on the given connection.
3026  */
ceph_con_send(struct ceph_connection * con,struct ceph_msg * msg)3027 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
3028 {
3029 	/* set src+dst */
3030 	msg->hdr.src = con->msgr->inst.name;
3031 	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
3032 	msg->needs_out_seq = true;
3033 
3034 	mutex_lock(&con->mutex);
3035 
3036 	if (con->state == CON_STATE_CLOSED) {
3037 		dout("con_send %p closed, dropping %p\n", con, msg);
3038 		ceph_msg_put(msg);
3039 		mutex_unlock(&con->mutex);
3040 		return;
3041 	}
3042 
3043 	msg_con_set(msg, con);
3044 
3045 	BUG_ON(!list_empty(&msg->list_head));
3046 	list_add_tail(&msg->list_head, &con->out_queue);
3047 	dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
3048 	     ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
3049 	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
3050 	     le32_to_cpu(msg->hdr.front_len),
3051 	     le32_to_cpu(msg->hdr.middle_len),
3052 	     le32_to_cpu(msg->hdr.data_len));
3053 
3054 	clear_standby(con);
3055 	mutex_unlock(&con->mutex);
3056 
3057 	/* if there wasn't anything waiting to send before, queue
3058 	 * new work */
3059 	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
3060 		queue_con(con);
3061 }
3062 EXPORT_SYMBOL(ceph_con_send);
3063 
3064 /*
3065  * Revoke a message that was previously queued for send
3066  */
ceph_msg_revoke(struct ceph_msg * msg)3067 void ceph_msg_revoke(struct ceph_msg *msg)
3068 {
3069 	struct ceph_connection *con = msg->con;
3070 
3071 	if (!con) {
3072 		dout("%s msg %p null con\n", __func__, msg);
3073 		return;		/* Message not in our possession */
3074 	}
3075 
3076 	mutex_lock(&con->mutex);
3077 	if (!list_empty(&msg->list_head)) {
3078 		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
3079 		list_del_init(&msg->list_head);
3080 		msg->hdr.seq = 0;
3081 
3082 		ceph_msg_put(msg);
3083 	}
3084 	if (con->out_msg == msg) {
3085 		BUG_ON(con->out_skip);
3086 		/* footer */
3087 		if (con->out_msg_done) {
3088 			con->out_skip += con_out_kvec_skip(con);
3089 		} else {
3090 			BUG_ON(!msg->data_length);
3091 			if (con->peer_features & CEPH_FEATURE_MSG_AUTH)
3092 				con->out_skip += sizeof(msg->footer);
3093 			else
3094 				con->out_skip += sizeof(msg->old_footer);
3095 		}
3096 		/* data, middle, front */
3097 		if (msg->data_length)
3098 			con->out_skip += msg->cursor.total_resid;
3099 		if (msg->middle)
3100 			con->out_skip += con_out_kvec_skip(con);
3101 		con->out_skip += con_out_kvec_skip(con);
3102 
3103 		dout("%s %p msg %p - was sending, will write %d skip %d\n",
3104 		     __func__, con, msg, con->out_kvec_bytes, con->out_skip);
3105 		msg->hdr.seq = 0;
3106 		con->out_msg = NULL;
3107 		ceph_msg_put(msg);
3108 	}
3109 
3110 	mutex_unlock(&con->mutex);
3111 }
3112 
3113 /*
3114  * Revoke a message that we may be reading data into
3115  */
ceph_msg_revoke_incoming(struct ceph_msg * msg)3116 void ceph_msg_revoke_incoming(struct ceph_msg *msg)
3117 {
3118 	struct ceph_connection *con = msg->con;
3119 
3120 	if (!con) {
3121 		dout("%s msg %p null con\n", __func__, msg);
3122 		return;		/* Message not in our possession */
3123 	}
3124 
3125 	mutex_lock(&con->mutex);
3126 	if (con->in_msg == msg) {
3127 		unsigned int front_len = le32_to_cpu(con->in_hdr.front_len);
3128 		unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len);
3129 		unsigned int data_len = le32_to_cpu(con->in_hdr.data_len);
3130 
3131 		/* skip rest of message */
3132 		dout("%s %p msg %p revoked\n", __func__, con, msg);
3133 		con->in_base_pos = con->in_base_pos -
3134 				sizeof(struct ceph_msg_header) -
3135 				front_len -
3136 				middle_len -
3137 				data_len -
3138 				sizeof(struct ceph_msg_footer);
3139 		ceph_msg_put(con->in_msg);
3140 		con->in_msg = NULL;
3141 		con->in_tag = CEPH_MSGR_TAG_READY;
3142 		con->in_seq++;
3143 	} else {
3144 		dout("%s %p in_msg %p msg %p no-op\n",
3145 		     __func__, con, con->in_msg, msg);
3146 	}
3147 	mutex_unlock(&con->mutex);
3148 }
3149 
3150 /*
3151  * Queue a keepalive byte to ensure the tcp connection is alive.
3152  */
ceph_con_keepalive(struct ceph_connection * con)3153 void ceph_con_keepalive(struct ceph_connection *con)
3154 {
3155 	dout("con_keepalive %p\n", con);
3156 	mutex_lock(&con->mutex);
3157 	clear_standby(con);
3158 	mutex_unlock(&con->mutex);
3159 	if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 &&
3160 	    con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
3161 		queue_con(con);
3162 }
3163 EXPORT_SYMBOL(ceph_con_keepalive);
3164 
ceph_con_keepalive_expired(struct ceph_connection * con,unsigned long interval)3165 bool ceph_con_keepalive_expired(struct ceph_connection *con,
3166 			       unsigned long interval)
3167 {
3168 	if (interval > 0 &&
3169 	    (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2)) {
3170 		struct timespec now = CURRENT_TIME;
3171 		struct timespec ts;
3172 		jiffies_to_timespec(interval, &ts);
3173 		ts = timespec_add(con->last_keepalive_ack, ts);
3174 		return timespec_compare(&now, &ts) >= 0;
3175 	}
3176 	return false;
3177 }
3178 
ceph_msg_data_create(enum ceph_msg_data_type type)3179 static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type)
3180 {
3181 	struct ceph_msg_data *data;
3182 
3183 	if (WARN_ON(!ceph_msg_data_type_valid(type)))
3184 		return NULL;
3185 
3186 	data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS);
3187 	if (data)
3188 		data->type = type;
3189 	INIT_LIST_HEAD(&data->links);
3190 
3191 	return data;
3192 }
3193 
ceph_msg_data_destroy(struct ceph_msg_data * data)3194 static void ceph_msg_data_destroy(struct ceph_msg_data *data)
3195 {
3196 	if (!data)
3197 		return;
3198 
3199 	WARN_ON(!list_empty(&data->links));
3200 	if (data->type == CEPH_MSG_DATA_PAGELIST)
3201 		ceph_pagelist_release(data->pagelist);
3202 	kmem_cache_free(ceph_msg_data_cache, data);
3203 }
3204 
ceph_msg_data_add_pages(struct ceph_msg * msg,struct page ** pages,size_t length,size_t alignment)3205 void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
3206 		size_t length, size_t alignment)
3207 {
3208 	struct ceph_msg_data *data;
3209 
3210 	BUG_ON(!pages);
3211 	BUG_ON(!length);
3212 
3213 	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
3214 	BUG_ON(!data);
3215 	data->pages = pages;
3216 	data->length = length;
3217 	data->alignment = alignment & ~PAGE_MASK;
3218 
3219 	list_add_tail(&data->links, &msg->data);
3220 	msg->data_length += length;
3221 }
3222 EXPORT_SYMBOL(ceph_msg_data_add_pages);
3223 
ceph_msg_data_add_pagelist(struct ceph_msg * msg,struct ceph_pagelist * pagelist)3224 void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
3225 				struct ceph_pagelist *pagelist)
3226 {
3227 	struct ceph_msg_data *data;
3228 
3229 	BUG_ON(!pagelist);
3230 	BUG_ON(!pagelist->length);
3231 
3232 	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
3233 	BUG_ON(!data);
3234 	data->pagelist = pagelist;
3235 
3236 	list_add_tail(&data->links, &msg->data);
3237 	msg->data_length += pagelist->length;
3238 }
3239 EXPORT_SYMBOL(ceph_msg_data_add_pagelist);
3240 
3241 #ifdef	CONFIG_BLOCK
ceph_msg_data_add_bio(struct ceph_msg * msg,struct bio * bio,size_t length)3242 void ceph_msg_data_add_bio(struct ceph_msg *msg, struct bio *bio,
3243 		size_t length)
3244 {
3245 	struct ceph_msg_data *data;
3246 
3247 	BUG_ON(!bio);
3248 
3249 	data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
3250 	BUG_ON(!data);
3251 	data->bio = bio;
3252 	data->bio_length = length;
3253 
3254 	list_add_tail(&data->links, &msg->data);
3255 	msg->data_length += length;
3256 }
3257 EXPORT_SYMBOL(ceph_msg_data_add_bio);
3258 #endif	/* CONFIG_BLOCK */
3259 
3260 /*
3261  * construct a new message with given type, size
3262  * the new msg has a ref count of 1.
3263  */
ceph_msg_new(int type,int front_len,gfp_t flags,bool can_fail)3264 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
3265 			      bool can_fail)
3266 {
3267 	struct ceph_msg *m;
3268 
3269 	m = kmem_cache_zalloc(ceph_msg_cache, flags);
3270 	if (m == NULL)
3271 		goto out;
3272 
3273 	m->hdr.type = cpu_to_le16(type);
3274 	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
3275 	m->hdr.front_len = cpu_to_le32(front_len);
3276 
3277 	INIT_LIST_HEAD(&m->list_head);
3278 	kref_init(&m->kref);
3279 	INIT_LIST_HEAD(&m->data);
3280 
3281 	/* front */
3282 	if (front_len) {
3283 		m->front.iov_base = ceph_kvmalloc(front_len, flags);
3284 		if (m->front.iov_base == NULL) {
3285 			dout("ceph_msg_new can't allocate %d bytes\n",
3286 			     front_len);
3287 			goto out2;
3288 		}
3289 	} else {
3290 		m->front.iov_base = NULL;
3291 	}
3292 	m->front_alloc_len = m->front.iov_len = front_len;
3293 
3294 	dout("ceph_msg_new %p front %d\n", m, front_len);
3295 	return m;
3296 
3297 out2:
3298 	ceph_msg_put(m);
3299 out:
3300 	if (!can_fail) {
3301 		pr_err("msg_new can't create type %d front %d\n", type,
3302 		       front_len);
3303 		WARN_ON(1);
3304 	} else {
3305 		dout("msg_new can't create type %d front %d\n", type,
3306 		     front_len);
3307 	}
3308 	return NULL;
3309 }
3310 EXPORT_SYMBOL(ceph_msg_new);
3311 
3312 /*
3313  * Allocate "middle" portion of a message, if it is needed and wasn't
3314  * allocated by alloc_msg.  This allows us to read a small fixed-size
3315  * per-type header in the front and then gracefully fail (i.e.,
3316  * propagate the error to the caller based on info in the front) when
3317  * the middle is too large.
3318  */
ceph_alloc_middle(struct ceph_connection * con,struct ceph_msg * msg)3319 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
3320 {
3321 	int type = le16_to_cpu(msg->hdr.type);
3322 	int middle_len = le32_to_cpu(msg->hdr.middle_len);
3323 
3324 	dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
3325 	     ceph_msg_type_name(type), middle_len);
3326 	BUG_ON(!middle_len);
3327 	BUG_ON(msg->middle);
3328 
3329 	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
3330 	if (!msg->middle)
3331 		return -ENOMEM;
3332 	return 0;
3333 }
3334 
3335 /*
3336  * Allocate a message for receiving an incoming message on a
3337  * connection, and save the result in con->in_msg.  Uses the
3338  * connection's private alloc_msg op if available.
3339  *
3340  * Returns 0 on success, or a negative error code.
3341  *
3342  * On success, if we set *skip = 1:
3343  *  - the next message should be skipped and ignored.
3344  *  - con->in_msg == NULL
3345  * or if we set *skip = 0:
3346  *  - con->in_msg is non-null.
3347  * On error (ENOMEM, EAGAIN, ...),
3348  *  - con->in_msg == NULL
3349  */
ceph_con_in_msg_alloc(struct ceph_connection * con,int * skip)3350 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3351 {
3352 	struct ceph_msg_header *hdr = &con->in_hdr;
3353 	int middle_len = le32_to_cpu(hdr->middle_len);
3354 	struct ceph_msg *msg;
3355 	int ret = 0;
3356 
3357 	BUG_ON(con->in_msg != NULL);
3358 	BUG_ON(!con->ops->alloc_msg);
3359 
3360 	mutex_unlock(&con->mutex);
3361 	msg = con->ops->alloc_msg(con, hdr, skip);
3362 	mutex_lock(&con->mutex);
3363 	if (con->state != CON_STATE_OPEN) {
3364 		if (msg)
3365 			ceph_msg_put(msg);
3366 		return -EAGAIN;
3367 	}
3368 	if (msg) {
3369 		BUG_ON(*skip);
3370 		msg_con_set(msg, con);
3371 		con->in_msg = msg;
3372 	} else {
3373 		/*
3374 		 * Null message pointer means either we should skip
3375 		 * this message or we couldn't allocate memory.  The
3376 		 * former is not an error.
3377 		 */
3378 		if (*skip)
3379 			return 0;
3380 
3381 		con->error_msg = "error allocating memory for incoming message";
3382 		return -ENOMEM;
3383 	}
3384 	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3385 
3386 	if (middle_len && !con->in_msg->middle) {
3387 		ret = ceph_alloc_middle(con, con->in_msg);
3388 		if (ret < 0) {
3389 			ceph_msg_put(con->in_msg);
3390 			con->in_msg = NULL;
3391 		}
3392 	}
3393 
3394 	return ret;
3395 }
3396 
3397 
3398 /*
3399  * Free a generically kmalloc'd message.
3400  */
ceph_msg_free(struct ceph_msg * m)3401 static void ceph_msg_free(struct ceph_msg *m)
3402 {
3403 	dout("%s %p\n", __func__, m);
3404 	kvfree(m->front.iov_base);
3405 	kmem_cache_free(ceph_msg_cache, m);
3406 }
3407 
ceph_msg_release(struct kref * kref)3408 static void ceph_msg_release(struct kref *kref)
3409 {
3410 	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
3411 	LIST_HEAD(data);
3412 	struct list_head *links;
3413 	struct list_head *next;
3414 
3415 	dout("%s %p\n", __func__, m);
3416 	WARN_ON(!list_empty(&m->list_head));
3417 
3418 	msg_con_set(m, NULL);
3419 
3420 	/* drop middle, data, if any */
3421 	if (m->middle) {
3422 		ceph_buffer_put(m->middle);
3423 		m->middle = NULL;
3424 	}
3425 
3426 	list_splice_init(&m->data, &data);
3427 	list_for_each_safe(links, next, &data) {
3428 		struct ceph_msg_data *data;
3429 
3430 		data = list_entry(links, struct ceph_msg_data, links);
3431 		list_del_init(links);
3432 		ceph_msg_data_destroy(data);
3433 	}
3434 	m->data_length = 0;
3435 
3436 	if (m->pool)
3437 		ceph_msgpool_put(m->pool, m);
3438 	else
3439 		ceph_msg_free(m);
3440 }
3441 
ceph_msg_get(struct ceph_msg * msg)3442 struct ceph_msg *ceph_msg_get(struct ceph_msg *msg)
3443 {
3444 	dout("%s %p (was %d)\n", __func__, msg,
3445 	     atomic_read(&msg->kref.refcount));
3446 	kref_get(&msg->kref);
3447 	return msg;
3448 }
3449 EXPORT_SYMBOL(ceph_msg_get);
3450 
ceph_msg_put(struct ceph_msg * msg)3451 void ceph_msg_put(struct ceph_msg *msg)
3452 {
3453 	dout("%s %p (was %d)\n", __func__, msg,
3454 	     atomic_read(&msg->kref.refcount));
3455 	kref_put(&msg->kref, ceph_msg_release);
3456 }
3457 EXPORT_SYMBOL(ceph_msg_put);
3458 
ceph_msg_dump(struct ceph_msg * msg)3459 void ceph_msg_dump(struct ceph_msg *msg)
3460 {
3461 	pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg,
3462 		 msg->front_alloc_len, msg->data_length);
3463 	print_hex_dump(KERN_DEBUG, "header: ",
3464 		       DUMP_PREFIX_OFFSET, 16, 1,
3465 		       &msg->hdr, sizeof(msg->hdr), true);
3466 	print_hex_dump(KERN_DEBUG, " front: ",
3467 		       DUMP_PREFIX_OFFSET, 16, 1,
3468 		       msg->front.iov_base, msg->front.iov_len, true);
3469 	if (msg->middle)
3470 		print_hex_dump(KERN_DEBUG, "middle: ",
3471 			       DUMP_PREFIX_OFFSET, 16, 1,
3472 			       msg->middle->vec.iov_base,
3473 			       msg->middle->vec.iov_len, true);
3474 	print_hex_dump(KERN_DEBUG, "footer: ",
3475 		       DUMP_PREFIX_OFFSET, 16, 1,
3476 		       &msg->footer, sizeof(msg->footer), true);
3477 }
3478 EXPORT_SYMBOL(ceph_msg_dump);
3479