messenger.c 84.6 KB
Newer Older
1
#include <linux/ceph/ceph_debug.h>
S
Sage Weil 已提交
2 3 4 5 6 7 8

#include <linux/crc32c.h>
#include <linux/ctype.h>
#include <linux/highmem.h>
#include <linux/inet.h>
#include <linux/kthread.h>
#include <linux/net.h>
9
#include <linux/nsproxy.h>
10
#include <linux/slab.h>
S
Sage Weil 已提交
11 12
#include <linux/socket.h>
#include <linux/string.h>
13
#ifdef	CONFIG_BLOCK
14
#include <linux/bio.h>
15
#endif	/* CONFIG_BLOCK */
N
Noah Watkins 已提交
16
#include <linux/dns_resolver.h>
S
Sage Weil 已提交
17 18
#include <net/tcp.h>

I
Ilya Dryomov 已提交
19
#include <linux/ceph/ceph_features.h>
20 21 22 23
#include <linux/ceph/libceph.h>
#include <linux/ceph/messenger.h>
#include <linux/ceph/decode.h>
#include <linux/ceph/pagelist.h>
24
#include <linux/export.h>
S
Sage Weil 已提交
25

26 27 28
#define list_entry_next(pos, member)					\
	list_entry(pos->member.next, typeof(*pos), member)

S
Sage Weil 已提交
29 30 31 32 33 34 35 36 37
/*
 * Ceph uses the messenger to exchange ceph_msg messages with other
 * hosts in the system.  The messenger provides ordered and reliable
 * delivery.  We tolerate TCP disconnects by reconnecting (with
 * exponential backoff) in the case of a fault (disconnection, bad
 * crc, protocol error).  Acks allow sent messages to be discarded by
 * the sender.
 */

A
Alex Elder 已提交
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56
/*
 * We track the state of the socket on a given connection using
 * values defined below.  The transition to a new socket state is
 * handled by a function which verifies we aren't coming from an
 * unexpected state.
 *
 *      --------
 *      | NEW* |  transient initial state
 *      --------
 *          | con_sock_state_init()
 *          v
 *      ----------
 *      | CLOSED |  initialized, but no socket (and no
 *      ----------  TCP connection)
 *       ^      \
 *       |       \ con_sock_state_connecting()
 *       |        ----------------------
 *       |                              \
 *       + con_sock_state_closed()       \
57 58 59 60 61 62 63 64 65 66 67
 *       |+---------------------------    \
 *       | \                          \    \
 *       |  -----------                \    \
 *       |  | CLOSING |  socket event;  \    \
 *       |  -----------  await close     \    \
 *       |       ^                        \   |
 *       |       |                         \  |
 *       |       + con_sock_state_closing() \ |
 *       |      / \                         | |
 *       |     /   ---------------          | |
 *       |    /                   \         v v
A
Alex Elder 已提交
68 69 70 71 72 73 74 75 76 77 78
 *       |   /                    --------------
 *       |  /    -----------------| CONNECTING |  socket created, TCP
 *       |  |   /                 --------------  connect initiated
 *       |  |   | con_sock_state_connected()
 *       |  |   v
 *      -------------
 *      | CONNECTED |  TCP connection established
 *      -------------
 *
 * State values for ceph_connection->sock_state; NEW is assumed to be 0.
 */
79 80 81 82 83 84 85

#define CON_SOCK_STATE_NEW		0	/* -> CLOSED */
#define CON_SOCK_STATE_CLOSED		1	/* -> CONNECTING */
#define CON_SOCK_STATE_CONNECTING	2	/* -> CONNECTED or -> CLOSING */
#define CON_SOCK_STATE_CONNECTED	3	/* -> CLOSING or -> CLOSED */
#define CON_SOCK_STATE_CLOSING		4	/* -> CLOSED */

86 87 88 89 90 91 92 93 94 95
/*
 * connection states
 */
#define CON_STATE_CLOSED        1  /* -> PREOPEN */
#define CON_STATE_PREOPEN       2  /* -> CONNECTING, CLOSED */
#define CON_STATE_CONNECTING    3  /* -> NEGOTIATING, CLOSED */
#define CON_STATE_NEGOTIATING   4  /* -> OPEN, CLOSED */
#define CON_STATE_OPEN          5  /* -> STANDBY, CLOSED */
#define CON_STATE_STANDBY       6  /* -> PREOPEN, CLOSED */

S
Sage Weil 已提交
96 97 98 99 100 101 102 103 104
/*
 * ceph_connection flag bits
 */
#define CON_FLAG_LOSSYTX           0  /* we can close channel or drop
				       * messages on errors */
#define CON_FLAG_KEEPALIVE_PENDING 1  /* we need to send a keepalive */
#define CON_FLAG_WRITE_PENDING	   2  /* we have data ready to send */
#define CON_FLAG_SOCK_CLOSED	   3  /* socket state changed to closed */
#define CON_FLAG_BACKOFF           4  /* need to retry queuing delayed work */
105

106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
static bool con_flag_valid(unsigned long con_flag)
{
	switch (con_flag) {
	case CON_FLAG_LOSSYTX:
	case CON_FLAG_KEEPALIVE_PENDING:
	case CON_FLAG_WRITE_PENDING:
	case CON_FLAG_SOCK_CLOSED:
	case CON_FLAG_BACKOFF:
		return true;
	default:
		return false;
	}
}

static void con_flag_clear(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	clear_bit(con_flag, &con->flags);
}

static void con_flag_set(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	set_bit(con_flag, &con->flags);
}

static bool con_flag_test(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_bit(con_flag, &con->flags);
}

static bool con_flag_test_and_clear(struct ceph_connection *con,
					unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_and_clear_bit(con_flag, &con->flags);
}

static bool con_flag_test_and_set(struct ceph_connection *con,
					unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_and_set_bit(con_flag, &con->flags);
}

157 158 159
/* Slab caches for frequently-allocated structures */

static struct kmem_cache	*ceph_msg_cache;
160
static struct kmem_cache	*ceph_msg_data_cache;
161

S
Sage Weil 已提交
162 163 164 165
/* static tag bytes (protocol control messages) */
static char tag_msg = CEPH_MSGR_TAG_MSG;
static char tag_ack = CEPH_MSGR_TAG_ACK;
static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
166
static char tag_keepalive2 = CEPH_MSGR_TAG_KEEPALIVE2;
S
Sage Weil 已提交
167

168 169 170 171
#ifdef CONFIG_LOCKDEP
static struct lock_class_key socket_class;
#endif

A
Alex Elder 已提交
172 173 174 175 176
/*
 * When skipping (ignoring) a block of input we read it into a "skip
 * buffer," which is this many bytes in size.
 */
#define SKIP_BUF_SIZE	1024
S
Sage Weil 已提交
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);
S
Sage Weil 已提交
182 183

/*
184 185
 * Nicely render a sockaddr as a string.  An array of formatted
 * strings is used, to approximate reentrancy.
S
Sage Weil 已提交
186
 */
187 188 189 190 191 192 193
#define ADDR_STR_COUNT_LOG	5	/* log2(# address strings in array) */
#define ADDR_STR_COUNT		(1 << ADDR_STR_COUNT_LOG)
#define ADDR_STR_COUNT_MASK	(ADDR_STR_COUNT - 1)
#define MAX_ADDR_STR_LEN	64	/* 54 is enough */

static char addr_str[ADDR_STR_COUNT][MAX_ADDR_STR_LEN];
static atomic_t addr_str_seq = ATOMIC_INIT(0);
S
Sage Weil 已提交
194

195 196
static struct page *zero_page;		/* used in certain error cases */

197
const char *ceph_pr_addr(const struct sockaddr_storage *ss)
S
Sage Weil 已提交
198 199 200
{
	int i;
	char *s;
A
Alex Elder 已提交
201 202
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
S
Sage Weil 已提交
203

204
	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
S
Sage Weil 已提交
205 206 207 208
	s = addr_str[i];

	switch (ss->ss_family) {
	case AF_INET:
A
Alex Elder 已提交
209 210
		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
			 ntohs(in4->sin_port));
S
Sage Weil 已提交
211 212 213
		break;

	case AF_INET6:
A
Alex Elder 已提交
214 215
		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
			 ntohs(in6->sin6_port));
S
Sage Weil 已提交
216 217 218
		break;

	default:
A
Alex Elder 已提交
219 220
		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
			 ss->ss_family);
S
Sage Weil 已提交
221 222 223 224
	}

	return s;
}
225
EXPORT_SYMBOL(ceph_pr_addr);
S
Sage Weil 已提交
226

227 228 229 230 231 232
static void encode_my_addr(struct ceph_messenger *msgr)
{
	memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
	ceph_encode_addr(&msgr->my_enc_addr);
}

S
Sage Weil 已提交
233 234 235
/*
 * work queue for all reading and writing to/from the socket.
 */
A
Alex Elder 已提交
236
static struct workqueue_struct *ceph_msgr_wq;
S
Sage Weil 已提交
237

238 239 240 241 242 243
static int ceph_msgr_slab_init(void)
{
	BUG_ON(ceph_msg_cache);
	ceph_msg_cache = kmem_cache_create("ceph_msg",
					sizeof (struct ceph_msg),
					__alignof__(struct ceph_msg), 0, NULL);
244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259

	if (!ceph_msg_cache)
		return -ENOMEM;

	BUG_ON(ceph_msg_data_cache);
	ceph_msg_data_cache = kmem_cache_create("ceph_msg_data",
					sizeof (struct ceph_msg_data),
					__alignof__(struct ceph_msg_data),
					0, NULL);
	if (ceph_msg_data_cache)
		return 0;

	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;

	return -ENOMEM;
260 261 262 263
}

static void ceph_msgr_slab_exit(void)
{
264 265 266 267
	BUG_ON(!ceph_msg_data_cache);
	kmem_cache_destroy(ceph_msg_data_cache);
	ceph_msg_data_cache = NULL;

268 269 270 271 272
	BUG_ON(!ceph_msg_cache);
	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;
}

A
Alex Elder 已提交
273
static void _ceph_msgr_exit(void)
274
{
A
Alex Elder 已提交
275
	if (ceph_msgr_wq) {
276
		destroy_workqueue(ceph_msgr_wq);
A
Alex Elder 已提交
277 278
		ceph_msgr_wq = NULL;
	}
279 280 281 282

	BUG_ON(zero_page == NULL);
	page_cache_release(zero_page);
	zero_page = NULL;
283 284

	ceph_msgr_slab_exit();
285 286
}

287
int ceph_msgr_init(void)
S
Sage Weil 已提交
288
{
289 290 291
	if (ceph_msgr_slab_init())
		return -ENOMEM;

292 293 294 295
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
	page_cache_get(zero_page);

296 297 298 299 300
	/*
	 * The number of active work items is limited by the number of
	 * connections, so leave @max_active at default.
	 */
	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_MEM_RECLAIM, 0);
301 302
	if (ceph_msgr_wq)
		return 0;
303

304 305
	pr_err("msgr_init failed to create workqueue\n");
	_ceph_msgr_exit();
306

307
	return -ENOMEM;
S
Sage Weil 已提交
308
}
309
EXPORT_SYMBOL(ceph_msgr_init);
S
Sage Weil 已提交
310 311 312

void ceph_msgr_exit(void)
{
313 314
	BUG_ON(ceph_msgr_wq == NULL);

315
	_ceph_msgr_exit();
S
Sage Weil 已提交
316
}
317
EXPORT_SYMBOL(ceph_msgr_exit);
S
Sage Weil 已提交
318

Y
Yehuda Sadeh 已提交
319
void ceph_msgr_flush(void)
320 321 322
{
	flush_workqueue(ceph_msgr_wq);
}
323
EXPORT_SYMBOL(ceph_msgr_flush);
324

325 326 327 328 329 330 331 332 333
/* Connection socket state transition functions */

static void con_sock_state_init(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	if (WARN_ON(old_state != CON_SOCK_STATE_NEW))
		printk("%s: unexpected old state %d\n", __func__, old_state);
334 335
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
336 337 338 339 340 341 342 343 344
}

static void con_sock_state_connecting(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTING);
	if (WARN_ON(old_state != CON_SOCK_STATE_CLOSED))
		printk("%s: unexpected old state %d\n", __func__, old_state);
345 346
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTING);
347 348 349 350 351 352 353 354 355
}

static void con_sock_state_connected(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTED);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING))
		printk("%s: unexpected old state %d\n", __func__, old_state);
356 357
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTED);
358 359 360 361 362 363 364 365 366 367 368
}

static void con_sock_state_closing(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSING);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING &&
			old_state != CON_SOCK_STATE_CONNECTED &&
			old_state != CON_SOCK_STATE_CLOSING))
		printk("%s: unexpected old state %d\n", __func__, old_state);
369 370
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSING);
371 372 373 374 375 376 377 378
}

static void con_sock_state_closed(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTED &&
379
		    old_state != CON_SOCK_STATE_CLOSING &&
380 381
		    old_state != CON_SOCK_STATE_CONNECTING &&
		    old_state != CON_SOCK_STATE_CLOSED))
382
		printk("%s: unexpected old state %d\n", __func__, old_state);
383 384
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
385
}
386

S
Sage Weil 已提交
387 388 389 390 391
/*
 * socket callback functions
 */

/* data available on socket, or listen socket received a connect */
392
static void ceph_sock_data_ready(struct sock *sk)
S
Sage Weil 已提交
393
{
A
Alex Elder 已提交
394
	struct ceph_connection *con = sk->sk_user_data;
395 396 397
	if (atomic_read(&con->msgr->stopping)) {
		return;
	}
A
Alex Elder 已提交
398

S
Sage Weil 已提交
399
	if (sk->sk_state != TCP_CLOSE_WAIT) {
A
Alex Elder 已提交
400
		dout("%s on %p state = %lu, queueing work\n", __func__,
S
Sage Weil 已提交
401 402 403 404 405 406
		     con, con->state);
		queue_con(con);
	}
}

/* socket has buffer space for writing */
A
Alex Elder 已提交
407
static void ceph_sock_write_space(struct sock *sk)
S
Sage Weil 已提交
408
{
A
Alex Elder 已提交
409
	struct ceph_connection *con = sk->sk_user_data;
S
Sage Weil 已提交
410

411 412
	/* only queue to workqueue if there is data we want to write,
	 * and there is sufficient space in the socket buffer to accept
A
Alex Elder 已提交
413
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
414 415 416 417
	 * doesn't get called again until try_write() fills the socket
	 * buffer. See net/ipv4/tcp_input.c:tcp_check_space()
	 * and net/core/stream.c:sk_stream_write_space().
	 */
418
	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
419
		if (sk_stream_is_writeable(sk)) {
A
Alex Elder 已提交
420
			dout("%s %p queueing write work\n", __func__, con);
421 422 423
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
S
Sage Weil 已提交
424
	} else {
A
Alex Elder 已提交
425
		dout("%s %p nothing to write\n", __func__, con);
S
Sage Weil 已提交
426 427 428 429
	}
}

/* socket's state has changed */
A
Alex Elder 已提交
430
static void ceph_sock_state_change(struct sock *sk)
S
Sage Weil 已提交
431
{
A
Alex Elder 已提交
432
	struct ceph_connection *con = sk->sk_user_data;
S
Sage Weil 已提交
433

A
Alex Elder 已提交
434
	dout("%s %p state = %lu sk_state = %u\n", __func__,
S
Sage Weil 已提交
435 436 437 438
	     con, con->state, sk->sk_state);

	switch (sk->sk_state) {
	case TCP_CLOSE:
A
Alex Elder 已提交
439
		dout("%s TCP_CLOSE\n", __func__);
S
Sage Weil 已提交
440
	case TCP_CLOSE_WAIT:
A
Alex Elder 已提交
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);
S
Sage Weil 已提交
445 446
		break;
	case TCP_ESTABLISHED:
A
Alex Elder 已提交
447
		dout("%s TCP_ESTABLISHED\n", __func__);
448
		con_sock_state_connected(con);
S
Sage Weil 已提交
449 450
		queue_con(con);
		break;
A
Alex Elder 已提交
451 452
	default:	/* Everything else is uninteresting */
		break;
S
Sage Weil 已提交
453 454 455 456 457 458 459 460 461 462
	}
}

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
A
Alex Elder 已提交
463
	sk->sk_user_data = con;
A
Alex Elder 已提交
464 465 466
	sk->sk_data_ready = ceph_sock_data_ready;
	sk->sk_write_space = ceph_sock_write_space;
	sk->sk_state_change = ceph_sock_state_change;
S
Sage Weil 已提交
467 468 469 470 471 472 473 474 475 476
}


/*
 * socket helpers
 */

/*
 * initiate connection to a remote socket.
 */
477
static int ceph_tcp_connect(struct ceph_connection *con)
S
Sage Weil 已提交
478
{
S
Sage Weil 已提交
479
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
S
Sage Weil 已提交
480 481 482 483
	struct socket *sock;
	int ret;

	BUG_ON(con->sock);
484
	ret = sock_create_kern(read_pnet(&con->msgr->net), paddr->ss_family,
485
			       SOCK_STREAM, IPPROTO_TCP, &sock);
S
Sage Weil 已提交
486
	if (ret)
487
		return ret;
488
	sock->sk->sk_allocation = GFP_NOFS;
S
Sage Weil 已提交
489

490 491 492 493
#ifdef CONFIG_LOCKDEP
	lockdep_set_class(&sock->sk->sk_lock, &socket_class);
#endif

S
Sage Weil 已提交
494 495
	set_sock_callbacks(sock, con);

496
	dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
497

498
	con_sock_state_connecting(con);
S
Sage Weil 已提交
499 500
	ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
				 O_NONBLOCK);
S
Sage Weil 已提交
501 502
	if (ret == -EINPROGRESS) {
		dout("connect %s EINPROGRESS sk_state = %u\n",
503
		     ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
504
		     sock->sk->sk_state);
505
	} else if (ret < 0) {
S
Sage Weil 已提交
506
		pr_err("connect %s error %d\n",
507
		       ceph_pr_addr(&con->peer_addr.in_addr), ret);
S
Sage Weil 已提交
508
		sock_release(sock);
509
		return ret;
510
	}
511

512
	if (ceph_test_opt(from_msgr(con->msgr), TCP_NODELAY)) {
C
Chaitanya Huilgol 已提交
513 514 515 516 517 518 519 520 521
		int optval = 1;

		ret = kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
					(char *)&optval, sizeof(optval));
		if (ret)
			pr_err("kernel_setsockopt(TCP_NODELAY) failed: %d",
			       ret);
	}

522
	con->sock = sock;
523
	return 0;
S
Sage Weil 已提交
524 525 526 527 528 529
}

static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
{
	struct kvec iov = {buf, len};
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
530
	int r;
S
Sage Weil 已提交
531

532 533 534 535
	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
536 537
}

538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553
static int ceph_tcp_recvpage(struct socket *sock, struct page *page,
		     int page_offset, size_t length)
{
	void *kaddr;
	int ret;

	BUG_ON(page_offset + length > PAGE_SIZE);

	kaddr = kmap(page);
	BUG_ON(!kaddr);
	ret = ceph_tcp_recvmsg(sock, kaddr + page_offset, length);
	kunmap(page);

	return ret;
}

S
Sage Weil 已提交
554 555 556 557 558 559 560 561
/*
 * write something.  @more is true if caller will be sending more data
 * shortly.
 */
static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
		     size_t kvlen, size_t len, int more)
{
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
562
	int r;
S
Sage Weil 已提交
563 564 565 566 567 568

	if (more)
		msg.msg_flags |= MSG_MORE;
	else
		msg.msg_flags |= MSG_EOR;  /* superfluous, but what the hell */

569 570 571 572
	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
573 574
}

575
static int __ceph_tcp_sendpage(struct socket *sock, struct page *page,
576
		     int offset, size_t size, bool more)
577 578 579 580 581 582 583 584 585 586 587
{
	int flags = MSG_DONTWAIT | MSG_NOSIGNAL | (more ? MSG_MORE : MSG_EOR);
	int ret;

	ret = kernel_sendpage(sock, page, offset, size, flags);
	if (ret == -EAGAIN)
		ret = 0;

	return ret;
}

588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605
static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
		     int offset, size_t size, bool more)
{
	int ret;
	struct kvec iov;

	/* sendpage cannot properly handle pages with page_count == 0,
	 * we need to fallback to sendmsg if that's the case */
	if (page_count(page) >= 1)
		return __ceph_tcp_sendpage(sock, page, offset, size, more);

	iov.iov_base = kmap(page) + offset;
	iov.iov_len = size;
	ret = ceph_tcp_sendmsg(sock, &iov, 1, size, more);
	kunmap(page);

	return ret;
}
S
Sage Weil 已提交
606 607 608 609 610 611

/*
 * Shutdown/close the socket for the given connection.
 */
static int con_close_socket(struct ceph_connection *con)
{
612
	int rc = 0;
S
Sage Weil 已提交
613 614

	dout("con_close_socket on %p sock %p\n", con, con->sock);
615 616 617 618 619
	if (con->sock) {
		rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
		sock_release(con->sock);
		con->sock = NULL;
	}
620 621

	/*
S
Sage Weil 已提交
622
	 * Forcibly clear the SOCK_CLOSED flag.  It gets set
623 624 625 626
	 * independent of the connection mutex, and we could have
	 * received a socket close event before we had the chance to
	 * shut the socket down.
	 */
627
	con_flag_clear(con, CON_FLAG_SOCK_CLOSED);
628

629
	con_sock_state_closed(con);
S
Sage Weil 已提交
630 631 632 633 634 635 636 637 638 639
	return rc;
}

/*
 * Reset a connection.  Discard all incoming and outgoing messages
 * and clear *_seq state.
 */
static void ceph_msg_remove(struct ceph_msg *msg)
{
	list_del_init(&msg->list_head);
640
	BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
641
	msg->con->ops->put(msg->con);
642 643
	msg->con = NULL;

S
Sage Weil 已提交
644 645 646 647 648 649 650 651 652 653 654 655 656 657 658
	ceph_msg_put(msg);
}
static void ceph_msg_remove_list(struct list_head *head)
{
	while (!list_empty(head)) {
		struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
							list_head);
		ceph_msg_remove(msg);
	}
}

static void reset_connection(struct ceph_connection *con)
{
	/* reset connection, out_queue, msg_ and connect_seq */
	/* discard existing out_queue and msg_seq */
659
	dout("reset_connection %p\n", con);
S
Sage Weil 已提交
660 661 662
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

663
	if (con->in_msg) {
664 665
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
666 667
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
S
Sage Weil 已提交
668
		con->ops->put(con);
669 670
	}

S
Sage Weil 已提交
671 672
	con->connect_seq = 0;
	con->out_seq = 0;
673 674 675 676
	if (con->out_msg) {
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
S
Sage Weil 已提交
677
	con->in_seq = 0;
678
	con->in_seq_acked = 0;
S
Sage Weil 已提交
679 680 681 682 683 684 685
}

/*
 * mark a peer down.  drop any open connections.
 */
void ceph_con_close(struct ceph_connection *con)
{
686
	mutex_lock(&con->mutex);
687 688
	dout("con_close %p peer %s\n", con,
	     ceph_pr_addr(&con->peer_addr.in_addr));
689
	con->state = CON_STATE_CLOSED;
690

691 692 693 694
	con_flag_clear(con, CON_FLAG_LOSSYTX);	/* so we retry next connect */
	con_flag_clear(con, CON_FLAG_KEEPALIVE_PENDING);
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
	con_flag_clear(con, CON_FLAG_BACKOFF);
695

S
Sage Weil 已提交
696
	reset_connection(con);
S
Sage Weil 已提交
697
	con->peer_global_seq = 0;
698
	cancel_con(con);
699
	con_close_socket(con);
700
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
701
}
702
EXPORT_SYMBOL(ceph_con_close);
S
Sage Weil 已提交
703 704 705 706

/*
 * Reopen a closed connection, with a new peer address.
 */
707 708 709
void ceph_con_open(struct ceph_connection *con,
		   __u8 entity_type, __u64 entity_num,
		   struct ceph_entity_addr *addr)
S
Sage Weil 已提交
710
{
711
	mutex_lock(&con->mutex);
712
	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
713

714
	WARN_ON(con->state != CON_STATE_CLOSED);
715
	con->state = CON_STATE_PREOPEN;
716

717 718 719
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

S
Sage Weil 已提交
720
	memcpy(&con->peer_addr, addr, sizeof(*addr));
721
	con->delay = 0;      /* reset backoff memory */
722
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
723 724
	queue_con(con);
}
725
EXPORT_SYMBOL(ceph_con_open);
S
Sage Weil 已提交
726

727 728 729 730 731 732 733 734
/*
 * return true if this connection ever successfully opened
 */
bool ceph_con_opened(struct ceph_connection *con)
{
	return con->connect_seq > 0;
}

S
Sage Weil 已提交
735 736 737
/*
 * initialize a new connection.
 */
738 739
void ceph_con_init(struct ceph_connection *con, void *private,
	const struct ceph_connection_operations *ops,
740
	struct ceph_messenger *msgr)
S
Sage Weil 已提交
741 742 743
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
744 745
	con->private = private;
	con->ops = ops;
S
Sage Weil 已提交
746
	con->msgr = msgr;
747 748 749

	con_sock_state_init(con);

750
	mutex_init(&con->mutex);
S
Sage Weil 已提交
751 752
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
753
	INIT_DELAYED_WORK(&con->work, ceph_con_workfn);
754

755
	con->state = CON_STATE_CLOSED;
S
Sage Weil 已提交
756
}
757
EXPORT_SYMBOL(ceph_con_init);
S
Sage Weil 已提交
758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775


/*
 * We maintain a global counter to order connection attempts.  Get
 * a unique seq greater than @gt.
 */
static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
{
	u32 ret;

	spin_lock(&msgr->global_seq_lock);
	if (msgr->global_seq < gt)
		msgr->global_seq = gt;
	ret = ++msgr->global_seq;
	spin_unlock(&msgr->global_seq_lock);
	return ret;
}

776
static void con_out_kvec_reset(struct ceph_connection *con)
777 778 779 780 781 782
{
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

783
static void con_out_kvec_add(struct ceph_connection *con,
784 785 786 787 788 789 790 791 792 793 794 795
				size_t size, void *data)
{
	int index;

	index = con->out_kvec_left;
	BUG_ON(index >= ARRAY_SIZE(con->out_kvec));

	con->out_kvec[index].iov_len = size;
	con->out_kvec[index].iov_base = data;
	con->out_kvec_left++;
	con->out_kvec_bytes += size;
}
S
Sage Weil 已提交
796

797
#ifdef CONFIG_BLOCK
798 799 800 801 802 803

/*
 * For a bio data item, a piece is whatever remains of the next
 * entry in the current bio iovec, or the first entry in the next
 * bio in the list.
 */
A
Alex Elder 已提交
804
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
805
					size_t length)
806
{
A
Alex Elder 已提交
807
	struct ceph_msg_data *data = cursor->data;
808 809 810 811 812 813 814
	struct bio *bio;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

	bio = data->bio;
	BUG_ON(!bio);

815
	cursor->resid = min(length, data->bio_length);
816
	cursor->bio = bio;
817 818 819
	cursor->bvec_iter = bio->bi_iter;
	cursor->last_piece =
		cursor->resid <= bio_iter_len(bio, cursor->bvec_iter);
820 821
}

A
Alex Elder 已提交
822
static struct page *ceph_msg_data_bio_next(struct ceph_msg_data_cursor *cursor,
823 824 825
						size_t *page_offset,
						size_t *length)
{
A
Alex Elder 已提交
826
	struct ceph_msg_data *data = cursor->data;
827
	struct bio *bio;
828
	struct bio_vec bio_vec;
829 830 831 832 833 834

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

	bio = cursor->bio;
	BUG_ON(!bio);

835
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
836

837
	*page_offset = (size_t) bio_vec.bv_offset;
838
	BUG_ON(*page_offset >= PAGE_SIZE);
839 840 841
	if (cursor->last_piece) /* pagelist offset is always 0 */
		*length = cursor->resid;
	else
842
		*length = (size_t) bio_vec.bv_len;
843
	BUG_ON(*length > cursor->resid);
844
	BUG_ON(*page_offset + *length > PAGE_SIZE);
845

846
	return bio_vec.bv_page;
847 848
}

A
Alex Elder 已提交
849 850
static bool ceph_msg_data_bio_advance(struct ceph_msg_data_cursor *cursor,
					size_t bytes)
851 852
{
	struct bio *bio;
853
	struct bio_vec bio_vec;
854

A
Alex Elder 已提交
855
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_BIO);
856 857 858 859

	bio = cursor->bio;
	BUG_ON(!bio);

860
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
861 862 863

	/* Advance the cursor offset */

864 865
	BUG_ON(cursor->resid < bytes);
	cursor->resid -= bytes;
866 867 868 869

	bio_advance_iter(bio, &cursor->bvec_iter, bytes);

	if (bytes < bio_vec.bv_len)
870 871 872 873
		return false;	/* more bytes to process in this segment */

	/* Move on to the next segment, and possibly the next bio */

874
	if (!cursor->bvec_iter.bi_size) {
875
		bio = bio->bi_next;
876 877 878 879 880 881
		cursor->bio = bio;
		if (bio)
			cursor->bvec_iter = bio->bi_iter;
		else
			memset(&cursor->bvec_iter, 0,
			       sizeof(cursor->bvec_iter));
882 883
	}

884 885 886 887
	if (!cursor->last_piece) {
		BUG_ON(!cursor->resid);
		BUG_ON(!bio);
		/* A short read is OK, so use <= rather than == */
888
		if (cursor->resid <= bio_iter_len(bio, cursor->bvec_iter))
889
			cursor->last_piece = true;
890
	}
891 892 893

	return true;
}
894
#endif /* CONFIG_BLOCK */
895

896 897 898 899
/*
 * For a page array, a piece comes from the first page in the array
 * that has not already been fully consumed.
 */
A
Alex Elder 已提交
900
static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor,
901
					size_t length)
902
{
A
Alex Elder 已提交
903
	struct ceph_msg_data *data = cursor->data;
904 905 906 907 908 909 910
	int page_count;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

	BUG_ON(!data->pages);
	BUG_ON(!data->length);

911
	cursor->resid = min(length, data->length);
912 913 914
	page_count = calc_pages_for(data->alignment, (u64)data->length);
	cursor->page_offset = data->alignment & ~PAGE_MASK;
	cursor->page_index = 0;
915 916 917
	BUG_ON(page_count > (int)USHRT_MAX);
	cursor->page_count = (unsigned short)page_count;
	BUG_ON(length > SIZE_MAX - cursor->page_offset);
918
	cursor->last_piece = cursor->page_offset + cursor->resid <= PAGE_SIZE;
919 920
}

A
Alex Elder 已提交
921 922 923
static struct page *
ceph_msg_data_pages_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length)
924
{
A
Alex Elder 已提交
925
	struct ceph_msg_data *data = cursor->data;
926 927 928 929 930 931 932

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

	BUG_ON(cursor->page_index >= cursor->page_count);
	BUG_ON(cursor->page_offset >= PAGE_SIZE);

	*page_offset = cursor->page_offset;
933
	if (cursor->last_piece)
934
		*length = cursor->resid;
935
	else
936 937 938 939 940
		*length = PAGE_SIZE - *page_offset;

	return data->pages[cursor->page_index];
}

A
Alex Elder 已提交
941
static bool ceph_msg_data_pages_advance(struct ceph_msg_data_cursor *cursor,
942 943
						size_t bytes)
{
A
Alex Elder 已提交
944
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_PAGES);
945 946 947 948 949 950

	BUG_ON(cursor->page_offset + bytes > PAGE_SIZE);

	/* Advance the cursor page offset */

	cursor->resid -= bytes;
951 952
	cursor->page_offset = (cursor->page_offset + bytes) & ~PAGE_MASK;
	if (!bytes || cursor->page_offset)
953 954
		return false;	/* more bytes to process in the current page */

955 956 957
	if (!cursor->resid)
		return false;   /* no more data */

958
	/* Move on to the next page; offset is already at 0 */
959 960 961

	BUG_ON(cursor->page_index >= cursor->page_count);
	cursor->page_index++;
962
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
963 964 965 966

	return true;
}

967
/*
968 969
 * For a pagelist, a piece is whatever remains to be consumed in the
 * first page in the list, or the front of the next page.
970
 */
A
Alex Elder 已提交
971 972
static void
ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
973
					size_t length)
974
{
A
Alex Elder 已提交
975
	struct ceph_msg_data *data = cursor->data;
976 977 978
	struct ceph_pagelist *pagelist;
	struct page *page;

979
	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
980 981 982

	pagelist = data->pagelist;
	BUG_ON(!pagelist);
983 984

	if (!length)
985 986 987 988 989
		return;		/* pagelist can be assigned but empty */

	BUG_ON(list_empty(&pagelist->head));
	page = list_first_entry(&pagelist->head, struct page, lru);

990
	cursor->resid = min(length, pagelist->length);
991 992
	cursor->page = page;
	cursor->offset = 0;
A
Alex Elder 已提交
993
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
994 995
}

A
Alex Elder 已提交
996 997 998
static struct page *
ceph_msg_data_pagelist_next(struct ceph_msg_data_cursor *cursor,
				size_t *page_offset, size_t *length)
999
{
A
Alex Elder 已提交
1000
	struct ceph_msg_data *data = cursor->data;
1001 1002 1003 1004 1005 1006 1007 1008
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

	pagelist = data->pagelist;
	BUG_ON(!pagelist);

	BUG_ON(!cursor->page);
1009
	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1010

1011
	/* offset of first page in pagelist is always 0 */
1012
	*page_offset = cursor->offset & ~PAGE_MASK;
1013
	if (cursor->last_piece)
1014 1015 1016
		*length = cursor->resid;
	else
		*length = PAGE_SIZE - *page_offset;
1017

A
Alex Elder 已提交
1018
	return cursor->page;
1019 1020
}

A
Alex Elder 已提交
1021
static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor,
1022
						size_t bytes)
1023
{
A
Alex Elder 已提交
1024
	struct ceph_msg_data *data = cursor->data;
1025 1026 1027 1028 1029 1030
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

	pagelist = data->pagelist;
	BUG_ON(!pagelist);
1031 1032

	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1033 1034 1035 1036
	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);

	/* Advance the cursor offset */

1037
	cursor->resid -= bytes;
1038
	cursor->offset += bytes;
1039
	/* offset of first page in pagelist is always 0 */
1040 1041 1042
	if (!bytes || cursor->offset & ~PAGE_MASK)
		return false;	/* more bytes to process in the current page */

1043 1044 1045
	if (!cursor->resid)
		return false;   /* no more data */

1046 1047 1048 1049
	/* Move on to the next page */

	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
	cursor->page = list_entry_next(cursor->page, lru);
1050
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1051 1052 1053 1054

	return true;
}

1055 1056 1057 1058 1059 1060 1061 1062
/*
 * Message data is handled (sent or received) in pieces, where each
 * piece resides on a single page.  The network layer might not
 * consume an entire piece at once.  A data item's cursor keeps
 * track of which piece is next to process and how much remains to
 * be processed in that piece.  It also tracks whether the current
 * piece is the last one in the data item.
 */
1063
static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
1064
{
1065
	size_t length = cursor->total_resid;
A
Alex Elder 已提交
1066 1067

	switch (cursor->data->type) {
1068
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1069
		ceph_msg_data_pagelist_cursor_init(cursor, length);
1070
		break;
1071
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1072
		ceph_msg_data_pages_cursor_init(cursor, length);
1073
		break;
1074 1075
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1076
		ceph_msg_data_bio_cursor_init(cursor, length);
1077
		break;
1078
#endif /* CONFIG_BLOCK */
1079
	case CEPH_MSG_DATA_NONE:
1080 1081 1082 1083
	default:
		/* BUG(); */
		break;
	}
A
Alex Elder 已提交
1084
	cursor->need_crc = true;
1085 1086
}

1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103
static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
{
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
	struct ceph_msg_data *data;

	BUG_ON(!length);
	BUG_ON(length > msg->data_length);
	BUG_ON(list_empty(&msg->data));

	cursor->data_head = &msg->data;
	cursor->total_resid = length;
	data = list_first_entry(&msg->data, struct ceph_msg_data, links);
	cursor->data = data;

	__ceph_msg_data_cursor_init(cursor);
}

1104 1105 1106 1107 1108
/*
 * Return the page containing the next piece to process for a given
 * data item, and supply the page offset and length of that piece.
 * Indicate whether this is the last piece in this data item.
 */
A
Alex Elder 已提交
1109 1110
static struct page *ceph_msg_data_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length,
1111 1112 1113 1114
					bool *last_piece)
{
	struct page *page;

A
Alex Elder 已提交
1115
	switch (cursor->data->type) {
1116
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1117
		page = ceph_msg_data_pagelist_next(cursor, page_offset, length);
1118
		break;
1119
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1120
		page = ceph_msg_data_pages_next(cursor, page_offset, length);
1121
		break;
1122 1123
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1124
		page = ceph_msg_data_bio_next(cursor, page_offset, length);
1125
		break;
1126
#endif /* CONFIG_BLOCK */
1127
	case CEPH_MSG_DATA_NONE:
1128 1129 1130 1131 1132 1133 1134 1135
	default:
		page = NULL;
		break;
	}
	BUG_ON(!page);
	BUG_ON(*page_offset + *length > PAGE_SIZE);
	BUG_ON(!*length);
	if (last_piece)
A
Alex Elder 已提交
1136
		*last_piece = cursor->last_piece;
1137 1138 1139 1140 1141 1142 1143 1144

	return page;
}

/*
 * Returns true if the result moves the cursor on to the next piece
 * of the data item.
 */
A
Alex Elder 已提交
1145 1146
static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor,
				size_t bytes)
1147 1148 1149
{
	bool new_piece;

1150
	BUG_ON(bytes > cursor->resid);
A
Alex Elder 已提交
1151
	switch (cursor->data->type) {
1152
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1153
		new_piece = ceph_msg_data_pagelist_advance(cursor, bytes);
1154
		break;
1155
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1156
		new_piece = ceph_msg_data_pages_advance(cursor, bytes);
1157
		break;
1158 1159
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1160
		new_piece = ceph_msg_data_bio_advance(cursor, bytes);
1161
		break;
1162
#endif /* CONFIG_BLOCK */
1163
	case CEPH_MSG_DATA_NONE:
1164 1165 1166 1167
	default:
		BUG();
		break;
	}
1168
	cursor->total_resid -= bytes;
1169

1170 1171 1172 1173 1174
	if (!cursor->resid && cursor->total_resid) {
		WARN_ON(!cursor->last_piece);
		BUG_ON(list_is_last(&cursor->data->links, cursor->data_head));
		cursor->data = list_entry_next(cursor->data, links);
		__ceph_msg_data_cursor_init(cursor);
A
Alex Elder 已提交
1175
		new_piece = true;
1176
	}
A
Alex Elder 已提交
1177
	cursor->need_crc = new_piece;
1178

1179 1180 1181
	return new_piece;
}

1182
static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
1183 1184
{
	BUG_ON(!msg);
1185
	BUG_ON(!data_len);
1186

1187
	/* Initialize data cursor */
1188

A
Alex Elder 已提交
1189
	ceph_msg_data_cursor_init(msg, (size_t)data_len);
1190 1191
}

S
Sage Weil 已提交
1192 1193 1194 1195
/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
1196
static void prepare_write_message_footer(struct ceph_connection *con)
S
Sage Weil 已提交
1197 1198
{
	struct ceph_msg *m = con->out_msg;
1199
	int v = con->out_kvec_left;
S
Sage Weil 已提交
1200

1201 1202
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

S
Sage Weil 已提交
1203 1204 1205
	dout("prepare_write_message_footer %p\n", con);
	con->out_kvec_is_msg = true;
	con->out_kvec[v].iov_base = &m->footer;
Y
Yan, Zheng 已提交
1206 1207
	if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
		if (con->ops->sign_message)
1208
			con->ops->sign_message(m);
Y
Yan, Zheng 已提交
1209 1210 1211 1212 1213 1214 1215 1216 1217
		else
			m->footer.sig = 0;
		con->out_kvec[v].iov_len = sizeof(m->footer);
		con->out_kvec_bytes += sizeof(m->footer);
	} else {
		m->old_footer.flags = m->footer.flags;
		con->out_kvec[v].iov_len = sizeof(m->old_footer);
		con->out_kvec_bytes += sizeof(m->old_footer);
	}
S
Sage Weil 已提交
1218 1219
	con->out_kvec_left++;
	con->out_more = m->more_to_follow;
1220
	con->out_msg_done = true;
S
Sage Weil 已提交
1221 1222 1223 1224 1225 1226 1227 1228
}

/*
 * Prepare headers for the next outgoing message.
 */
static void prepare_write_message(struct ceph_connection *con)
{
	struct ceph_msg *m;
1229
	u32 crc;
S
Sage Weil 已提交
1230

1231
	con_out_kvec_reset(con);
S
Sage Weil 已提交
1232
	con->out_kvec_is_msg = true;
1233
	con->out_msg_done = false;
S
Sage Weil 已提交
1234 1235 1236 1237 1238

	/* Sneak an ack in there first?  If we can get it into the same
	 * TCP packet that's a good thing. */
	if (con->in_seq > con->in_seq_acked) {
		con->in_seq_acked = con->in_seq;
1239
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
S
Sage Weil 已提交
1240
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1241
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1242
			&con->out_temp_ack);
S
Sage Weil 已提交
1243 1244
	}

1245
	BUG_ON(list_empty(&con->out_queue));
1246
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
1247
	con->out_msg = m;
1248
	BUG_ON(m->con != con);
1249 1250 1251 1252

	/* put message on sent list */
	ceph_msg_get(m);
	list_move_tail(&m->list_head, &con->out_sent);
S
Sage Weil 已提交
1253

1254 1255 1256 1257 1258 1259 1260 1261
	/*
	 * only assign outgoing seq # if we haven't sent this message
	 * yet.  if it is requeued, resend with it's original seq.
	 */
	if (m->needs_out_seq) {
		m->hdr.seq = cpu_to_le64(++con->out_seq);
		m->needs_out_seq = false;
	}
1262
	WARN_ON(m->data_length != le32_to_cpu(m->hdr.data_len));
S
Sage Weil 已提交
1263

1264
	dout("prepare_write_message %p seq %lld type %d len %d+%d+%zd\n",
S
Sage Weil 已提交
1265 1266
	     m, con->out_seq, le16_to_cpu(m->hdr.type),
	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
1267
	     m->data_length);
S
Sage Weil 已提交
1268 1269 1270
	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

	/* tag + hdr + front + middle */
1271 1272 1273
	con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
	con_out_kvec_add(con, sizeof (m->hdr), &m->hdr);
	con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
1274

S
Sage Weil 已提交
1275
	if (m->middle)
1276
		con_out_kvec_add(con, m->middle->vec.iov_len,
1277
			m->middle->vec.iov_base);
S
Sage Weil 已提交
1278 1279

	/* fill in crc (except data pages), footer */
1280 1281
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
1282
	con->out_msg->footer.flags = 0;
1283 1284 1285 1286 1287 1288 1289 1290

	crc = crc32c(0, m->front.iov_base, m->front.iov_len);
	con->out_msg->footer.front_crc = cpu_to_le32(crc);
	if (m->middle) {
		crc = crc32c(0, m->middle->vec.iov_base,
				m->middle->vec.iov_len);
		con->out_msg->footer.middle_crc = cpu_to_le32(crc);
	} else
S
Sage Weil 已提交
1291
		con->out_msg->footer.middle_crc = 0;
1292
	dout("%s front_crc %u middle_crc %u\n", __func__,
S
Sage Weil 已提交
1293 1294 1295 1296
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));

	/* is there a data payload? */
1297
	con->out_msg->footer.data_crc = 0;
1298 1299
	if (m->data_length) {
		prepare_message_data(con->out_msg, m->data_length);
1300 1301
		con->out_more = 1;  /* data + footer will follow */
	} else {
S
Sage Weil 已提交
1302
		/* no, queue up footer too and be done */
1303
		prepare_write_message_footer(con);
1304
	}
S
Sage Weil 已提交
1305

1306
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317
}

/*
 * Prepare an ack.
 */
static void prepare_write_ack(struct ceph_connection *con)
{
	dout("prepare_write_ack %p %llu -> %llu\n", con,
	     con->in_seq_acked, con->in_seq);
	con->in_seq_acked = con->in_seq;

1318
	con_out_kvec_reset(con);
1319

1320
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1321

S
Sage Weil 已提交
1322
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1323
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1324 1325
				&con->out_temp_ack);

S
Sage Weil 已提交
1326
	con->out_more = 1;  /* more will follow.. eventually.. */
1327
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1328 1329
}

1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347
/*
 * Prepare to share the seq during handshake
 */
static void prepare_write_seq(struct ceph_connection *con)
{
	dout("prepare_write_seq %p %llu -> %llu\n", con,
	     con->in_seq_acked, con->in_seq);
	con->in_seq_acked = con->in_seq;

	con_out_kvec_reset(con);

	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
			 &con->out_temp_ack);

	con_flag_set(con, CON_FLAG_WRITE_PENDING);
}

S
Sage Weil 已提交
1348 1349 1350 1351 1352 1353
/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
1354
	con_out_kvec_reset(con);
1355
	if (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2) {
1356 1357
		struct timespec now = CURRENT_TIME;

1358
		con_out_kvec_add(con, sizeof(tag_keepalive2), &tag_keepalive2);
1359 1360 1361
		ceph_encode_timespec(&con->out_temp_keepalive2, &now);
		con_out_kvec_add(con, sizeof(con->out_temp_keepalive2),
				 &con->out_temp_keepalive2);
1362 1363 1364
	} else {
		con_out_kvec_add(con, sizeof(tag_keepalive), &tag_keepalive);
	}
1365
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1366 1367 1368 1369 1370 1371
}

/*
 * Connection negotiation.
 */

1372 1373
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
1374
{
1375
	struct ceph_auth_handshake *auth;
1376 1377 1378 1379

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;
1380
		return NULL;
1381 1382 1383
	}

	/* Can't hold the mutex while getting authorizer */
1384
	mutex_unlock(&con->mutex);
1385
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1386
	mutex_lock(&con->mutex);
1387

1388
	if (IS_ERR(auth))
1389
		return auth;
1390
	if (con->state != CON_STATE_NEGOTIATING)
1391
		return ERR_PTR(-EAGAIN);
1392

1393 1394
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1395
	return auth;
1396 1397
}

S
Sage Weil 已提交
1398 1399 1400
/*
 * We connected to a peer and are saying hello.
 */
1401
static void prepare_write_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1402
{
1403 1404
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1405
					&con->msgr->my_enc_addr);
1406 1407

	con->out_more = 0;
1408
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1409 1410
}

1411
static int prepare_write_connect(struct ceph_connection *con)
1412
{
1413
	unsigned int global_seq = get_global_seq(con->msgr, 0);
S
Sage Weil 已提交
1414
	int proto;
1415
	int auth_proto;
1416
	struct ceph_auth_handshake *auth;
S
Sage Weil 已提交
1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433

	switch (con->peer_name.type) {
	case CEPH_ENTITY_TYPE_MON:
		proto = CEPH_MONC_PROTOCOL;
		break;
	case CEPH_ENTITY_TYPE_OSD:
		proto = CEPH_OSDC_PROTOCOL;
		break;
	case CEPH_ENTITY_TYPE_MDS:
		proto = CEPH_MDSC_PROTOCOL;
		break;
	default:
		BUG();
	}

	dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
	     con->connect_seq, global_seq, proto);
1434

1435 1436
	con->out_connect.features =
	    cpu_to_le64(from_msgr(con->msgr)->supported_features);
S
Sage Weil 已提交
1437 1438 1439 1440 1441 1442
	con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
	con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
	con->out_connect.global_seq = cpu_to_le32(global_seq);
	con->out_connect.protocol_version = cpu_to_le32(proto);
	con->out_connect.flags = 0;

1443 1444
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
1445 1446
	if (IS_ERR(auth))
		return PTR_ERR(auth);
1447

1448
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1449 1450 1451
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

1452
	con_out_kvec_add(con, sizeof (con->out_connect),
1453 1454
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
1455
		con_out_kvec_add(con, auth->authorizer_buf_len,
1456
					auth->authorizer_buf);
1457

S
Sage Weil 已提交
1458
	con->out_more = 0;
1459
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1460

1461
	return 0;
S
Sage Weil 已提交
1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483
}

/*
 * write as much of pending kvecs to the socket as we can.
 *  1 -> done
 *  0 -> socket full, but more to do
 * <0 -> error
 */
static int write_partial_kvec(struct ceph_connection *con)
{
	int ret;

	dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
	while (con->out_kvec_bytes > 0) {
		ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
				       con->out_kvec_left, con->out_kvec_bytes,
				       con->out_more);
		if (ret <= 0)
			goto out;
		con->out_kvec_bytes -= ret;
		if (con->out_kvec_bytes == 0)
			break;            /* done */
1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495

		/* account for full iov entries consumed */
		while (ret >= con->out_kvec_cur->iov_len) {
			BUG_ON(!con->out_kvec_left);
			ret -= con->out_kvec_cur->iov_len;
			con->out_kvec_cur++;
			con->out_kvec_left--;
		}
		/* and for a partially-consumed entry */
		if (ret) {
			con->out_kvec_cur->iov_len -= ret;
			con->out_kvec_cur->iov_base += ret;
S
Sage Weil 已提交
1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506
		}
	}
	con->out_kvec_left = 0;
	con->out_kvec_is_msg = false;
	ret = 1;
out:
	dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
	     con->out_kvec_bytes, con->out_kvec_left, ret);
	return ret;  /* done! */
}

1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519
static u32 ceph_crc32c_page(u32 crc, struct page *page,
				unsigned int page_offset,
				unsigned int length)
{
	char *kaddr;

	kaddr = kmap(page);
	BUG_ON(kaddr == NULL);
	crc = crc32c(crc, kaddr + page_offset, length);
	kunmap(page);

	return crc;
}
S
Sage Weil 已提交
1520 1521 1522 1523 1524 1525 1526
/*
 * Write as much message data payload as we can.  If we finish, queue
 * up the footer.
 *  1 -> done, footer is now queued in out_kvec[].
 *  0 -> socket full, but more to do
 * <0 -> error
 */
1527
static int write_partial_message_data(struct ceph_connection *con)
S
Sage Weil 已提交
1528 1529
{
	struct ceph_msg *msg = con->out_msg;
A
Alex Elder 已提交
1530
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1531
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
A
Alex Elder 已提交
1532
	u32 crc;
S
Sage Weil 已提交
1533

A
Alex Elder 已提交
1534
	dout("%s %p msg %p\n", __func__, con, msg);
S
Sage Weil 已提交
1535

1536
	if (list_empty(&msg->data))
1537 1538
		return -EINVAL;

1539 1540 1541 1542 1543 1544 1545 1546
	/*
	 * Iterate through each page that contains data to be
	 * written, and send as much as possible for each.
	 *
	 * If we are calculating the data crc (the default), we will
	 * need to map the page.  If we have no pages, they have
	 * been revoked, so use the zero page.
	 */
A
Alex Elder 已提交
1547
	crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0;
1548
	while (cursor->resid) {
1549
		struct page *page;
1550 1551
		size_t page_offset;
		size_t length;
1552
		bool last_piece;
1553
		bool need_crc;
A
Alex Elder 已提交
1554
		int ret;
1555

1556 1557
		page = ceph_msg_data_next(cursor, &page_offset, &length,
					  &last_piece);
1558
		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1559
					length, !last_piece);
A
Alex Elder 已提交
1560 1561 1562
		if (ret <= 0) {
			if (do_datacrc)
				msg->footer.data_crc = cpu_to_le32(crc);
S
Sage Weil 已提交
1563

A
Alex Elder 已提交
1564 1565
			return ret;
		}
1566 1567
		if (do_datacrc && cursor->need_crc)
			crc = ceph_crc32c_page(crc, page, page_offset, length);
1568
		need_crc = ceph_msg_data_advance(cursor, (size_t)ret);
S
Sage Weil 已提交
1569 1570
	}

1571
	dout("%s %p msg %p done\n", __func__, con, msg);
S
Sage Weil 已提交
1572 1573

	/* prepare and queue up footer, too */
A
Alex Elder 已提交
1574 1575 1576
	if (do_datacrc)
		msg->footer.data_crc = cpu_to_le32(crc);
	else
1577
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1578
	con_out_kvec_reset(con);
1579
	prepare_write_message_footer(con);
A
Alex Elder 已提交
1580 1581

	return 1;	/* must return > 0 to indicate success */
S
Sage Weil 已提交
1582 1583 1584 1585 1586 1587 1588 1589 1590 1591
}

/*
 * write some zeros
 */
static int write_partial_skip(struct ceph_connection *con)
{
	int ret;

	while (con->out_skip > 0) {
1592
		size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
S
Sage Weil 已提交
1593

1594
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
S
Sage Weil 已提交
1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1607 1608 1609 1610 1611 1612
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624
static void prepare_read_connect(struct ceph_connection *con)
{
	dout("prepare_read_connect %p\n", con);
	con->in_base_pos = 0;
}

static void prepare_read_ack(struct ceph_connection *con)
{
	dout("prepare_read_ack %p\n", con);
	con->in_base_pos = 0;
}

1625 1626 1627 1628 1629 1630 1631
static void prepare_read_seq(struct ceph_connection *con)
{
	dout("prepare_read_seq %p\n", con);
	con->in_base_pos = 0;
	con->in_tag = CEPH_MSGR_TAG_SEQ;
}

S
Sage Weil 已提交
1632 1633 1634 1635 1636 1637 1638
static void prepare_read_tag(struct ceph_connection *con)
{
	dout("prepare_read_tag %p\n", con);
	con->in_base_pos = 0;
	con->in_tag = CEPH_MSGR_TAG_READY;
}

1639 1640 1641 1642 1643 1644
static void prepare_read_keepalive_ack(struct ceph_connection *con)
{
	dout("prepare_read_keepalive_ack %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658
/*
 * Prepare to read a message.
 */
static int prepare_read_message(struct ceph_connection *con)
{
	dout("prepare_read_message %p\n", con);
	BUG_ON(con->in_msg != NULL);
	con->in_base_pos = 0;
	con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
	return 0;
}


static int read_partial(struct ceph_connection *con,
1659
			int end, int size, void *object)
S
Sage Weil 已提交
1660
{
1661 1662
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
S
Sage Weil 已提交
1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675
		int have = size - left;
		int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
		if (ret <= 0)
			return ret;
		con->in_base_pos += ret;
	}
	return 1;
}


/*
 * Read all or part of the connect-side handshake on a new connection
 */
1676
static int read_partial_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1677
{
1678 1679 1680
	int size;
	int end;
	int ret;
S
Sage Weil 已提交
1681

1682
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
S
Sage Weil 已提交
1683 1684

	/* peer's banner */
1685 1686 1687
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
S
Sage Weil 已提交
1688 1689
	if (ret <= 0)
		goto out;
1690 1691 1692 1693

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
S
Sage Weil 已提交
1694 1695
	if (ret <= 0)
		goto out;
1696 1697 1698 1699

	size = sizeof (con->peer_addr_for_me);
	end += size;
	ret = read_partial(con, end, size, &con->peer_addr_for_me);
S
Sage Weil 已提交
1700 1701
	if (ret <= 0)
		goto out;
1702

1703 1704 1705 1706 1707 1708
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1709 1710 1711
	int size;
	int end;
	int ret;
1712 1713 1714

	dout("read_partial_connect %p at %d\n", con, con->in_base_pos);

1715 1716 1717
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
S
Sage Weil 已提交
1718 1719
	if (ret <= 0)
		goto out;
1720 1721 1722 1723

	size = le32_to_cpu(con->in_reply.authorizer_len);
	end += size;
	ret = read_partial(con, end, size, con->auth_reply_buf);
1724 1725
	if (ret <= 0)
		goto out;
S
Sage Weil 已提交
1726

1727 1728 1729
	dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
	     con, (int)con->in_reply.tag,
	     le32_to_cpu(con->in_reply.connect_seq),
S
Sage Weil 已提交
1730 1731 1732
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1733

S
Sage Weil 已提交
1734 1735 1736 1737 1738 1739 1740 1741
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1742
		pr_err("connect to %s got bad banner\n",
1743
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1744 1745 1746 1747 1748 1749 1750 1751
		con->error_msg = "protocol error, bad banner";
		return -1;
	}
	return 0;
}

static bool addr_is_blank(struct sockaddr_storage *ss)
{
1752 1753 1754
	struct in_addr *addr = &((struct sockaddr_in *)ss)->sin_addr;
	struct in6_addr *addr6 = &((struct sockaddr_in6 *)ss)->sin6_addr;

S
Sage Weil 已提交
1755 1756
	switch (ss->ss_family) {
	case AF_INET:
1757
		return addr->s_addr == htonl(INADDR_ANY);
S
Sage Weil 已提交
1758
	case AF_INET6:
1759 1760 1761
		return ipv6_addr_any(addr6);
	default:
		return true;
S
Sage Weil 已提交
1762 1763 1764 1765 1766 1767 1768
	}
}

static int addr_port(struct sockaddr_storage *ss)
{
	switch (ss->ss_family) {
	case AF_INET:
S
Sage Weil 已提交
1769
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
S
Sage Weil 已提交
1770
	case AF_INET6:
S
Sage Weil 已提交
1771
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
S
Sage Weil 已提交
1772 1773 1774 1775 1776 1777 1778 1779 1780
	}
	return 0;
}

static void addr_set_port(struct sockaddr_storage *ss, int p)
{
	switch (ss->ss_family) {
	case AF_INET:
		((struct sockaddr_in *)ss)->sin_port = htons(p);
1781
		break;
S
Sage Weil 已提交
1782 1783
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1784
		break;
S
Sage Weil 已提交
1785 1786 1787
	}
}

N
Noah Watkins 已提交
1788 1789 1790 1791 1792 1793
/*
 * Unlike other *_pton function semantics, zero indicates success.
 */
static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
		char delim, const char **ipend)
{
A
Alex Elder 已提交
1794 1795
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
N
Noah Watkins 已提交
1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882

	memset(ss, 0, sizeof(*ss));

	if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
		ss->ss_family = AF_INET;
		return 0;
	}

	if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
		ss->ss_family = AF_INET6;
		return 0;
	}

	return -EINVAL;
}

/*
 * Extract hostname string and resolve using kernel DNS facility.
 */
#ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
static int ceph_dns_resolve_name(const char *name, size_t namelen,
		struct sockaddr_storage *ss, char delim, const char **ipend)
{
	const char *end, *delim_p;
	char *colon_p, *ip_addr = NULL;
	int ip_len, ret;

	/*
	 * The end of the hostname occurs immediately preceding the delimiter or
	 * the port marker (':') where the delimiter takes precedence.
	 */
	delim_p = memchr(name, delim, namelen);
	colon_p = memchr(name, ':', namelen);

	if (delim_p && colon_p)
		end = delim_p < colon_p ? delim_p : colon_p;
	else if (!delim_p && colon_p)
		end = colon_p;
	else {
		end = delim_p;
		if (!end) /* case: hostname:/ */
			end = name + namelen;
	}

	if (end <= name)
		return -EINVAL;

	/* do dns_resolve upcall */
	ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
	if (ip_len > 0)
		ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
	else
		ret = -ESRCH;

	kfree(ip_addr);

	*ipend = end;

	pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
			ret, ret ? "failed" : ceph_pr_addr(ss));

	return ret;
}
#else
static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
		struct sockaddr_storage *ss, char delim, const char **ipend)
{
	return -EINVAL;
}
#endif

/*
 * Parse a server name (IP or hostname). If a valid IP address is not found
 * then try to extract a hostname to resolve using userspace DNS upcall.
 */
static int ceph_parse_server_name(const char *name, size_t namelen,
			struct sockaddr_storage *ss, char delim, const char **ipend)
{
	int ret;

	ret = ceph_pton(name, namelen, ss, delim, ipend);
	if (ret)
		ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);

	return ret;
}

S
Sage Weil 已提交
1883 1884 1885 1886 1887 1888 1889 1890
/*
 * Parse an ip[:port] list into an addr array.  Use the default
 * monitor port if a port isn't specified.
 */
int ceph_parse_ips(const char *c, const char *end,
		   struct ceph_entity_addr *addr,
		   int max_count, int *count)
{
N
Noah Watkins 已提交
1891
	int i, ret = -EINVAL;
S
Sage Weil 已提交
1892 1893 1894 1895 1896 1897 1898
	const char *p = c;

	dout("parse_ips on '%.*s'\n", (int)(end-c), c);
	for (i = 0; i < max_count; i++) {
		const char *ipend;
		struct sockaddr_storage *ss = &addr[i].in_addr;
		int port;
S
Sage Weil 已提交
1899 1900 1901 1902 1903 1904
		char delim = ',';

		if (*p == '[') {
			delim = ']';
			p++;
		}
S
Sage Weil 已提交
1905

N
Noah Watkins 已提交
1906 1907
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
S
Sage Weil 已提交
1908
			goto bad;
N
Noah Watkins 已提交
1909 1910
		ret = -EINVAL;

S
Sage Weil 已提交
1911 1912
		p = ipend;

S
Sage Weil 已提交
1913 1914 1915 1916 1917 1918 1919 1920
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

S
Sage Weil 已提交
1921 1922 1923 1924 1925 1926 1927 1928
		/* port? */
		if (p < end && *p == ':') {
			port = 0;
			p++;
			while (p < end && *p >= '0' && *p <= '9') {
				port = (port * 10) + (*p - '0');
				p++;
			}
1929 1930 1931
			if (port == 0)
				port = CEPH_MON_PORT;
			else if (port > 65535)
S
Sage Weil 已提交
1932 1933 1934 1935 1936 1937 1938
				goto bad;
		} else {
			port = CEPH_MON_PORT;
		}

		addr_set_port(ss, port);

1939
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
S
Sage Weil 已提交
1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955

		if (p == end)
			break;
		if (*p != ',')
			goto bad;
		p++;
	}

	if (p != end)
		goto bad;

	if (count)
		*count = i + 1;
	return 0;

bad:
S
Sage Weil 已提交
1956
	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
N
Noah Watkins 已提交
1957
	return ret;
S
Sage Weil 已提交
1958
}
1959
EXPORT_SYMBOL(ceph_parse_ips);
S
Sage Weil 已提交
1960

1961
static int process_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1962
{
1963
	dout("process_banner on %p\n", con);
S
Sage Weil 已提交
1964 1965 1966 1967

	if (verify_hello(con) < 0)
		return -1;

1968 1969 1970
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

S
Sage Weil 已提交
1971 1972 1973 1974 1975
	/*
	 * Make sure the other end is who we wanted.  note that the other
	 * end may not yet know their ip address, so if it's 0.0.0.0, give
	 * them the benefit of the doubt.
	 */
S
Sage Weil 已提交
1976 1977
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
S
Sage Weil 已提交
1978 1979
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
1980 1981 1982 1983 1984
		pr_warn("wrong peer, want %s/%d, got %s/%d\n",
			ceph_pr_addr(&con->peer_addr.in_addr),
			(int)le32_to_cpu(con->peer_addr.nonce),
			ceph_pr_addr(&con->actual_peer_addr.in_addr),
			(int)le32_to_cpu(con->actual_peer_addr.nonce));
1985
		con->error_msg = "wrong peer at address";
S
Sage Weil 已提交
1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998
		return -1;
	}

	/*
	 * did we learn our address?
	 */
	if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
		int port = addr_port(&con->msgr->inst.addr.in_addr);

		memcpy(&con->msgr->inst.addr.in_addr,
		       &con->peer_addr_for_me.in_addr,
		       sizeof(con->peer_addr_for_me.in_addr));
		addr_set_port(&con->msgr->inst.addr.in_addr, port);
1999
		encode_my_addr(con->msgr);
2000
		dout("process_banner learned my addr is %s\n",
2001
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
S
Sage Weil 已提交
2002 2003
	}

2004 2005 2006 2007 2008
	return 0;
}

static int process_connect(struct ceph_connection *con)
{
2009 2010
	u64 sup_feat = from_msgr(con->msgr)->supported_features;
	u64 req_feat = from_msgr(con->msgr)->required_features;
I
Ilya Dryomov 已提交
2011 2012
	u64 server_feat = ceph_sanitize_features(
				le64_to_cpu(con->in_reply.features));
2013
	int ret;
2014

2015 2016
	dout("process_connect on %p tag %d\n", con, (int)con->in_tag);

S
Sage Weil 已提交
2017
	switch (con->in_reply.tag) {
2018 2019 2020 2021
	case CEPH_MSGR_TAG_FEATURES:
		pr_err("%s%lld %s feature set mismatch,"
		       " my %llx < server's %llx, missing %llx\n",
		       ENTITY_NAME(con->peer_name),
2022
		       ceph_pr_addr(&con->peer_addr.in_addr),
2023 2024
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
2025
		reset_connection(con);
2026 2027
		return -1;

S
Sage Weil 已提交
2028 2029 2030 2031
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
2032
		       ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
2033 2034 2035
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
2036
		reset_connection(con);
S
Sage Weil 已提交
2037 2038
		return -1;

2039 2040 2041 2042 2043 2044 2045 2046
	case CEPH_MSGR_TAG_BADAUTHORIZER:
		con->auth_retry++;
		dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
		     con->auth_retry);
		if (con->auth_retry == 2) {
			con->error_msg = "connect authorization failure";
			return -1;
		}
2047
		con_out_kvec_reset(con);
2048
		ret = prepare_write_connect(con);
2049 2050
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2051
		prepare_read_connect(con);
2052
		break;
S
Sage Weil 已提交
2053 2054 2055 2056 2057 2058 2059 2060 2061 2062

	case CEPH_MSGR_TAG_RESETSESSION:
		/*
		 * If we connected with a large connect_seq but the peer
		 * has no record of a session with us (no connection, or
		 * connect_seq == 0), they will send RESETSESION to indicate
		 * that they must have reset their session, and may have
		 * dropped messages.
		 */
		dout("process_connect got RESET peer seq %u\n",
S
Sage Weil 已提交
2063
		     le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2064 2065
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
2066
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2067
		reset_connection(con);
2068
		con_out_kvec_reset(con);
2069 2070 2071
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2072 2073 2074
		prepare_read_connect(con);

		/* Tell ceph about it. */
2075
		mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2076 2077 2078
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
2079
		mutex_lock(&con->mutex);
2080
		if (con->state != CON_STATE_NEGOTIATING)
2081
			return -EAGAIN;
S
Sage Weil 已提交
2082 2083 2084 2085 2086 2087 2088
		break;

	case CEPH_MSGR_TAG_RETRY_SESSION:
		/*
		 * If we sent a smaller connect_seq than the peer has, try
		 * again with a larger value.
		 */
S
Sage Weil 已提交
2089
		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
S
Sage Weil 已提交
2090
		     le32_to_cpu(con->out_connect.connect_seq),
S
Sage Weil 已提交
2091 2092
		     le32_to_cpu(con->in_reply.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2093
		con_out_kvec_reset(con);
2094 2095 2096
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2097 2098 2099 2100 2101 2102 2103 2104
		prepare_read_connect(con);
		break;

	case CEPH_MSGR_TAG_RETRY_GLOBAL:
		/*
		 * If we sent a smaller global_seq than the peer has, try
		 * again with a larger value.
		 */
2105
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
S
Sage Weil 已提交
2106
		     con->peer_global_seq,
S
Sage Weil 已提交
2107
		     le32_to_cpu(con->in_reply.global_seq));
S
Sage Weil 已提交
2108
		get_global_seq(con->msgr,
S
Sage Weil 已提交
2109
			       le32_to_cpu(con->in_reply.global_seq));
2110
		con_out_kvec_reset(con);
2111 2112 2113
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2114 2115 2116
		prepare_read_connect(con);
		break;

2117
	case CEPH_MSGR_TAG_SEQ:
S
Sage Weil 已提交
2118
	case CEPH_MSGR_TAG_READY:
2119 2120 2121 2122
		if (req_feat & ~server_feat) {
			pr_err("%s%lld %s protocol feature mismatch,"
			       " my required %llx > server's %llx, need %llx\n",
			       ENTITY_NAME(con->peer_name),
2123
			       ceph_pr_addr(&con->peer_addr.in_addr),
2124 2125
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
2126
			reset_connection(con);
2127 2128
			return -1;
		}
2129

2130
		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2131
		con->state = CON_STATE_OPEN;
2132
		con->auth_retry = 0;    /* we authenticated; clear flag */
S
Sage Weil 已提交
2133 2134
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
2135
		con->peer_features = server_feat;
S
Sage Weil 已提交
2136 2137 2138 2139 2140 2141
		dout("process_connect got READY gseq %d cseq %d (%d)\n",
		     con->peer_global_seq,
		     le32_to_cpu(con->in_reply.connect_seq),
		     con->connect_seq);
		WARN_ON(con->connect_seq !=
			le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2142 2143

		if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
2144
			con_flag_set(con, CON_FLAG_LOSSYTX);
S
Sage Weil 已提交
2145

2146
		con->delay = 0;      /* reset backoff memory */
S
Sage Weil 已提交
2147

2148 2149 2150 2151 2152 2153
		if (con->in_reply.tag == CEPH_MSGR_TAG_SEQ) {
			prepare_write_seq(con);
			prepare_read_seq(con);
		} else {
			prepare_read_tag(con);
		}
S
Sage Weil 已提交
2154 2155 2156 2157 2158 2159 2160 2161 2162
		break;

	case CEPH_MSGR_TAG_WAIT:
		/*
		 * If there is a connection race (we are opening
		 * connections to each other), one of us may just have
		 * to WAIT.  This shouldn't happen if we are the
		 * client.
		 */
S
Sage Weil 已提交
2163 2164
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
S
Sage Weil 已提交
2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178

	default:
		con->error_msg = "protocol error, garbage tag during connect";
		return -1;
	}
	return 0;
}


/*
 * read (part of) an ack
 */
static int read_partial_ack(struct ceph_connection *con)
{
2179 2180
	int size = sizeof (con->in_temp_ack);
	int end = size;
S
Sage Weil 已提交
2181

2182
	return read_partial(con, end, size, &con->in_temp_ack);
S
Sage Weil 已提交
2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201
}

/*
 * We can finally discard anything that's been acked.
 */
static void process_ack(struct ceph_connection *con)
{
	struct ceph_msg *m;
	u64 ack = le64_to_cpu(con->in_temp_ack);
	u64 seq;

	while (!list_empty(&con->out_sent)) {
		m = list_first_entry(&con->out_sent, struct ceph_msg,
				     list_head);
		seq = le64_to_cpu(m->hdr.seq);
		if (seq > ack)
			break;
		dout("got ack for seq %llu type %d at %p\n", seq,
		     le16_to_cpu(m->hdr.type), m);
2202
		m->ack_stamp = jiffies;
S
Sage Weil 已提交
2203 2204 2205 2206 2207 2208
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}


2209
static int read_partial_message_section(struct ceph_connection *con,
S
Sage Weil 已提交
2210 2211
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
2212
{
2213
	int ret, left;
2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225

	BUG_ON(!section);

	while (section->iov_len < sec_len) {
		BUG_ON(section->iov_base == NULL);
		left = sec_len - section->iov_len;
		ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
				       section->iov_len, left);
		if (ret <= 0)
			return ret;
		section->iov_len += ret;
	}
2226 2227
	if (section->iov_len == sec_len)
		*crc = crc32c(0, section->iov_base, section->iov_len);
S
Sage Weil 已提交
2228

2229 2230
	return 1;
}
S
Sage Weil 已提交
2231

2232 2233 2234
static int read_partial_msg_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->in_msg;
A
Alex Elder 已提交
2235
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
2236
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
A
Alex Elder 已提交
2237 2238 2239
	struct page *page;
	size_t page_offset;
	size_t length;
A
Alex Elder 已提交
2240
	u32 crc = 0;
2241 2242 2243
	int ret;

	BUG_ON(!msg);
2244
	if (list_empty(&msg->data))
2245
		return -EIO;
2246

A
Alex Elder 已提交
2247 2248
	if (do_datacrc)
		crc = con->in_data_crc;
2249
	while (cursor->resid) {
2250
		page = ceph_msg_data_next(cursor, &page_offset, &length, NULL);
A
Alex Elder 已提交
2251
		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
A
Alex Elder 已提交
2252 2253 2254 2255
		if (ret <= 0) {
			if (do_datacrc)
				con->in_data_crc = crc;

A
Alex Elder 已提交
2256
			return ret;
A
Alex Elder 已提交
2257
		}
A
Alex Elder 已提交
2258 2259

		if (do_datacrc)
A
Alex Elder 已提交
2260
			crc = ceph_crc32c_page(crc, page, page_offset, ret);
2261
		(void) ceph_msg_data_advance(cursor, (size_t)ret);
2262
	}
A
Alex Elder 已提交
2263 2264
	if (do_datacrc)
		con->in_data_crc = crc;
2265 2266 2267 2268

	return 1;	/* must return > 0 to indicate success */
}

S
Sage Weil 已提交
2269 2270 2271
/*
 * read (part of) a message.
 */
A
Alex Elder 已提交
2272 2273
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);

S
Sage Weil 已提交
2274 2275 2276
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
2277 2278
	int size;
	int end;
S
Sage Weil 已提交
2279
	int ret;
2280
	unsigned int front_len, middle_len, data_len;
2281
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
Y
Yan, Zheng 已提交
2282
	bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
2283
	u64 seq;
2284
	u32 crc;
S
Sage Weil 已提交
2285 2286 2287 2288

	dout("read_partial_message con %p msg %p\n", con, m);

	/* header */
2289 2290 2291
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
2292 2293
	if (ret <= 0)
		return ret;
2294 2295 2296

	crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
	if (cpu_to_le32(crc) != con->in_hdr.crc) {
2297
		pr_err("read_partial_message bad hdr crc %u != expected %u\n",
2298 2299 2300 2301
		       crc, con->in_hdr.crc);
		return -EBADMSG;
	}

S
Sage Weil 已提交
2302 2303 2304 2305
	front_len = le32_to_cpu(con->in_hdr.front_len);
	if (front_len > CEPH_MSG_MAX_FRONT_LEN)
		return -EIO;
	middle_len = le32_to_cpu(con->in_hdr.middle_len);
2306
	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
S
Sage Weil 已提交
2307 2308 2309 2310 2311
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

2312 2313 2314
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
S
Sage Weil 已提交
2315
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2316
			ENTITY_NAME(con->peer_name),
2317
			ceph_pr_addr(&con->peer_addr.in_addr),
2318 2319 2320 2321 2322 2323 2324 2325 2326
			seq, con->in_seq + 1);
		con->in_base_pos = -front_len - middle_len - data_len -
			sizeof(m->footer);
		con->in_tag = CEPH_MSGR_TAG_READY;
		return 0;
	} else if ((s64)seq - (s64)con->in_seq > 1) {
		pr_err("read_partial_message bad seq %lld expected %lld\n",
		       seq, con->in_seq + 1);
		con->error_msg = "bad message sequence # for incoming message";
2327
		return -EBADE;
2328 2329
	}

S
Sage Weil 已提交
2330 2331
	/* allocate message? */
	if (!con->in_msg) {
2332 2333
		int skip = 0;

S
Sage Weil 已提交
2334
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2335
		     front_len, data_len);
2336 2337 2338
		ret = ceph_con_in_msg_alloc(con, &skip);
		if (ret < 0)
			return ret;
2339 2340

		BUG_ON(!con->in_msg ^ skip);
2341
		if (skip) {
S
Sage Weil 已提交
2342
			/* skip this message */
2343
			dout("alloc_msg said skip message\n");
S
Sage Weil 已提交
2344 2345 2346
			con->in_base_pos = -front_len - middle_len - data_len -
				sizeof(m->footer);
			con->in_tag = CEPH_MSGR_TAG_READY;
2347
			con->in_seq++;
S
Sage Weil 已提交
2348 2349
			return 0;
		}
2350

2351
		BUG_ON(!con->in_msg);
2352
		BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
2353 2354
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
2355 2356
		if (m->middle)
			m->middle->vec.iov_len = 0;
2357

2358
		/* prepare for data payload, if any */
2359

2360
		if (data_len)
2361
			prepare_message_data(con->in_msg, data_len);
S
Sage Weil 已提交
2362 2363 2364
	}

	/* front */
2365 2366 2367 2368
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
S
Sage Weil 已提交
2369 2370

	/* middle */
2371
	if (m->middle) {
S
Sage Weil 已提交
2372 2373
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
2374
						   &con->in_middle_crc);
S
Sage Weil 已提交
2375 2376 2377 2378 2379
		if (ret <= 0)
			return ret;
	}

	/* (page) data */
2380 2381 2382 2383
	if (data_len) {
		ret = read_partial_msg_data(con);
		if (ret <= 0)
			return ret;
S
Sage Weil 已提交
2384 2385 2386
	}

	/* footer */
Y
Yan, Zheng 已提交
2387 2388 2389 2390 2391
	if (need_sign)
		size = sizeof(m->footer);
	else
		size = sizeof(m->old_footer);

2392 2393
	end += size;
	ret = read_partial(con, end, size, &m->footer);
2394 2395 2396
	if (ret <= 0)
		return ret;

Y
Yan, Zheng 已提交
2397 2398 2399 2400 2401
	if (!need_sign) {
		m->footer.flags = m->old_footer.flags;
		m->footer.sig = 0;
	}

S
Sage Weil 已提交
2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416
	dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
	     m, front_len, m->footer.front_crc, middle_len,
	     m->footer.middle_crc, data_len, m->footer.data_crc);

	/* crc ok? */
	if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
		pr_err("read_partial_message %p front crc %u != exp. %u\n",
		       m, con->in_front_crc, m->footer.front_crc);
		return -EBADMSG;
	}
	if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
		pr_err("read_partial_message %p middle crc %u != exp %u\n",
		       m, con->in_middle_crc, m->footer.middle_crc);
		return -EBADMSG;
	}
2417
	if (do_datacrc &&
S
Sage Weil 已提交
2418 2419 2420 2421 2422 2423 2424
	    (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
	    con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
		pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
		       con->in_data_crc, le32_to_cpu(m->footer.data_crc));
		return -EBADMSG;
	}

Y
Yan, Zheng 已提交
2425
	if (need_sign && con->ops->check_message_signature &&
2426
	    con->ops->check_message_signature(m)) {
Y
Yan, Zheng 已提交
2427 2428 2429 2430
		pr_err("read_partial_message %p signature check failed\n", m);
		return -EBADMSG;
	}

S
Sage Weil 已提交
2431 2432 2433 2434 2435 2436 2437 2438 2439 2440
	return 1; /* done! */
}

/*
 * Process message.  This happens in the worker thread.  The callback should
 * be careful not to do anything that waits on other incoming messages or it
 * may deadlock.
 */
static void process_message(struct ceph_connection *con)
{
S
Sage Weil 已提交
2441
	struct ceph_msg *msg;
S
Sage Weil 已提交
2442

2443 2444
	BUG_ON(con->in_msg->con != con);
	con->in_msg->con = NULL;
S
Sage Weil 已提交
2445
	msg = con->in_msg;
S
Sage Weil 已提交
2446
	con->in_msg = NULL;
S
Sage Weil 已提交
2447
	con->ops->put(con);
S
Sage Weil 已提交
2448 2449 2450

	/* if first message, set peer_name */
	if (con->peer_name.type == 0)
2451
		con->peer_name = msg->hdr.src;
S
Sage Weil 已提交
2452 2453

	con->in_seq++;
2454
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2455 2456 2457

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
2458
	     ENTITY_NAME(msg->hdr.src),
S
Sage Weil 已提交
2459 2460 2461 2462 2463 2464
	     le16_to_cpu(msg->hdr.type),
	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
	     le32_to_cpu(msg->hdr.front_len),
	     le32_to_cpu(msg->hdr.data_len),
	     con->in_front_crc, con->in_middle_crc, con->in_data_crc);
	con->ops->dispatch(con, msg);
2465 2466

	mutex_lock(&con->mutex);
S
Sage Weil 已提交
2467 2468
}

2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479
static int read_keepalive_ack(struct ceph_connection *con)
{
	struct ceph_timespec ceph_ts;
	size_t size = sizeof(ceph_ts);
	int ret = read_partial(con, size, size, &ceph_ts);
	if (ret <= 0)
		return ret;
	ceph_decode_timespec(&con->last_keepalive_ack, &ceph_ts);
	prepare_read_tag(con);
	return 1;
}
S
Sage Weil 已提交
2480 2481 2482 2483 2484 2485 2486 2487 2488

/*
 * Write something to the socket.  Called in a worker thread when the
 * socket appears to be writeable and we have something ready to send.
 */
static int try_write(struct ceph_connection *con)
{
	int ret = 1;

2489
	dout("try_write start %p state %lu\n", con, con->state);
S
Sage Weil 已提交
2490 2491 2492 2493 2494

more:
	dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);

	/* open the socket first? */
2495 2496 2497
	if (con->state == CON_STATE_PREOPEN) {
		BUG_ON(con->sock);
		con->state = CON_STATE_CONNECTING;
2498

2499
		con_out_kvec_reset(con);
2500
		prepare_write_banner(con);
2501
		prepare_read_banner(con);
S
Sage Weil 已提交
2502

2503
		BUG_ON(con->in_msg);
S
Sage Weil 已提交
2504 2505 2506
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
2507 2508
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
S
Sage Weil 已提交
2509 2510 2511 2512 2513 2514 2515 2516 2517 2518
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
	if (con->out_skip) {
		ret = write_partial_skip(con);
		if (ret <= 0)
2519
			goto out;
S
Sage Weil 已提交
2520 2521 2522 2523
	}
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
		if (ret <= 0)
2524
			goto out;
S
Sage Weil 已提交
2525 2526 2527 2528
	}

	/* msg pages? */
	if (con->out_msg) {
2529 2530 2531 2532 2533 2534
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

2535
		ret = write_partial_message_data(con);
S
Sage Weil 已提交
2536 2537 2538
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2539
			goto out;
S
Sage Weil 已提交
2540
		if (ret < 0) {
2541
			dout("try_write write_partial_message_data err %d\n",
S
Sage Weil 已提交
2542
			     ret);
2543
			goto out;
S
Sage Weil 已提交
2544 2545 2546
		}
	}

2547
do_next:
2548
	if (con->state == CON_STATE_OPEN) {
2549 2550 2551 2552
		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
			prepare_write_keepalive(con);
			goto more;
		}
S
Sage Weil 已提交
2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564
		/* is anything else pending? */
		if (!list_empty(&con->out_queue)) {
			prepare_write_message(con);
			goto more;
		}
		if (con->in_seq > con->in_seq_acked) {
			prepare_write_ack(con);
			goto more;
		}
	}

	/* Nothing to do! */
2565
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
2566 2567 2568
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2569
	dout("try_write done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581
	return ret;
}



/*
 * Read what we can from the socket.
 */
static int try_read(struct ceph_connection *con)
{
	int ret = -1;

2582 2583 2584 2585 2586
more:
	dout("try_read start on %p state %lu\n", con, con->state);
	if (con->state != CON_STATE_CONNECTING &&
	    con->state != CON_STATE_NEGOTIATING &&
	    con->state != CON_STATE_OPEN)
S
Sage Weil 已提交
2587 2588
		return 0;

2589
	BUG_ON(!con->sock);
2590

S
Sage Weil 已提交
2591 2592
	dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
	     con->in_base_pos);
2593

2594
	if (con->state == CON_STATE_CONNECTING) {
2595 2596 2597
		dout("try_read connecting\n");
		ret = read_partial_banner(con);
		if (ret <= 0)
2598
			goto out;
2599 2600 2601 2602
		ret = process_banner(con);
		if (ret < 0)
			goto out;

2603
		con->state = CON_STATE_NEGOTIATING;
2604

2605 2606 2607 2608 2609
		/*
		 * Received banner is good, exchange connection info.
		 * Do not reset out_kvec, as sending our banner raced
		 * with receiving peer banner after connect completed.
		 */
2610 2611 2612 2613 2614 2615
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
		prepare_read_connect(con);

		/* Send connection info before awaiting response */
2616 2617 2618
		goto out;
	}

2619
	if (con->state == CON_STATE_NEGOTIATING) {
2620
		dout("try_read negotiating\n");
S
Sage Weil 已提交
2621 2622 2623
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2624 2625 2626
		ret = process_connect(con);
		if (ret < 0)
			goto out;
S
Sage Weil 已提交
2627 2628 2629
		goto more;
	}

2630
	WARN_ON(con->state != CON_STATE_OPEN);
2631

S
Sage Weil 已提交
2632 2633 2634 2635 2636 2637
	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
A
Alex Elder 已提交
2638 2639 2640
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

S
Sage Weil 已提交
2641 2642 2643
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2644
			goto out;
S
Sage Weil 已提交
2645 2646 2647 2648 2649 2650 2651 2652 2653 2654
		con->in_base_pos += ret;
		if (con->in_base_pos)
			goto more;
	}
	if (con->in_tag == CEPH_MSGR_TAG_READY) {
		/*
		 * what's next?
		 */
		ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
		if (ret <= 0)
2655
			goto out;
S
Sage Weil 已提交
2656 2657 2658 2659 2660 2661 2662 2663
		dout("try_read got tag %d\n", (int)con->in_tag);
		switch (con->in_tag) {
		case CEPH_MSGR_TAG_MSG:
			prepare_read_message(con);
			break;
		case CEPH_MSGR_TAG_ACK:
			prepare_read_ack(con);
			break;
2664 2665 2666
		case CEPH_MSGR_TAG_KEEPALIVE2_ACK:
			prepare_read_keepalive_ack(con);
			break;
S
Sage Weil 已提交
2667
		case CEPH_MSGR_TAG_CLOSE:
2668 2669
			con_close_socket(con);
			con->state = CON_STATE_CLOSED;
2670
			goto out;
S
Sage Weil 已提交
2671 2672 2673 2674 2675 2676 2677 2678 2679
		default:
			goto bad_tag;
		}
	}
	if (con->in_tag == CEPH_MSGR_TAG_MSG) {
		ret = read_partial_message(con);
		if (ret <= 0) {
			switch (ret) {
			case -EBADMSG:
2680
				con->error_msg = "bad crc/signature";
2681 2682
				/* fall through */
			case -EBADE:
S
Sage Weil 已提交
2683
				ret = -EIO;
2684
				break;
S
Sage Weil 已提交
2685 2686
			case -EIO:
				con->error_msg = "io error";
2687
				break;
S
Sage Weil 已提交
2688
			}
2689
			goto out;
S
Sage Weil 已提交
2690 2691 2692 2693
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
2694 2695
		if (con->state == CON_STATE_OPEN)
			prepare_read_tag(con);
S
Sage Weil 已提交
2696 2697
		goto more;
	}
2698 2699 2700 2701 2702 2703
	if (con->in_tag == CEPH_MSGR_TAG_ACK ||
	    con->in_tag == CEPH_MSGR_TAG_SEQ) {
		/*
		 * the final handshake seq exchange is semantically
		 * equivalent to an ACK
		 */
S
Sage Weil 已提交
2704 2705
		ret = read_partial_ack(con);
		if (ret <= 0)
2706
			goto out;
S
Sage Weil 已提交
2707 2708 2709
		process_ack(con);
		goto more;
	}
2710 2711 2712 2713 2714 2715
	if (con->in_tag == CEPH_MSGR_TAG_KEEPALIVE2_ACK) {
		ret = read_keepalive_ack(con);
		if (ret <= 0)
			goto out;
		goto more;
	}
S
Sage Weil 已提交
2716 2717

out:
2718
	dout("try_read done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2719 2720 2721 2722 2723 2724 2725 2726 2727 2728 2729
	return ret;

bad_tag:
	pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
	con->error_msg = "protocol error, garbage tag";
	ret = -1;
	goto out;
}


/*
A
Alex Elder 已提交
2730 2731 2732
 * Atomically queue work on a connection after the specified delay.
 * Bump @con reference to avoid races with connection teardown.
 * Returns 0 if work was queued, or an error code otherwise.
S
Sage Weil 已提交
2733
 */
A
Alex Elder 已提交
2734
static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
S
Sage Weil 已提交
2735 2736
{
	if (!con->ops->get(con)) {
A
Alex Elder 已提交
2737 2738
		dout("%s %p ref count 0\n", __func__, con);
		return -ENOENT;
S
Sage Weil 已提交
2739 2740
	}

A
Alex Elder 已提交
2741 2742
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
		dout("%s %p - already queued\n", __func__, con);
S
Sage Weil 已提交
2743
		con->ops->put(con);
A
Alex Elder 已提交
2744
		return -EBUSY;
S
Sage Weil 已提交
2745
	}
A
Alex Elder 已提交
2746 2747 2748 2749 2750 2751 2752 2753

	dout("%s %p %lu\n", __func__, con, delay);
	return 0;
}

static void queue_con(struct ceph_connection *con)
{
	(void) queue_con_delay(con, 0);
S
Sage Weil 已提交
2754 2755
}

2756 2757 2758 2759 2760 2761 2762 2763
static void cancel_con(struct ceph_connection *con)
{
	if (cancel_delayed_work(&con->work)) {
		dout("%s %p\n", __func__, con);
		con->ops->put(con);
	}
}

2764 2765
static bool con_sock_closed(struct ceph_connection *con)
{
2766
	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781
		return false;

#define CASE(x)								\
	case CON_STATE_ ## x:						\
		con->error_msg = "socket closed (con state " #x ")";	\
		break;

	switch (con->state) {
	CASE(CLOSED);
	CASE(PREOPEN);
	CASE(CONNECTING);
	CASE(NEGOTIATING);
	CASE(OPEN);
	CASE(STANDBY);
	default:
2782
		pr_warn("%s con %p unrecognized state %lu\n",
2783 2784 2785 2786 2787 2788 2789 2790 2791 2792
			__func__, con, con->state);
		con->error_msg = "unrecognized con state";
		BUG();
		break;
	}
#undef CASE

	return true;
}

2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810
static bool con_backoff(struct ceph_connection *con)
{
	int ret;

	if (!con_flag_test_and_clear(con, CON_FLAG_BACKOFF))
		return false;

	ret = queue_con_delay(con, round_jiffies_relative(con->delay));
	if (ret) {
		dout("%s: con %p FAILED to back off %lu\n", __func__,
			con, con->delay);
		BUG_ON(ret == -ENOENT);
		con_flag_set(con, CON_FLAG_BACKOFF);
	}

	return true;
}

2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823 2824 2825 2826 2827
/* Finish fault handling; con->mutex must *not* be held here */

static void con_fault_finish(struct ceph_connection *con)
{
	/*
	 * in case we faulted due to authentication, invalidate our
	 * current tickets so that we can get new ones.
	 */
	if (con->auth_retry && con->ops->invalidate_authorizer) {
		dout("calling invalidate_authorizer()\n");
		con->ops->invalidate_authorizer(con);
	}

	if (con->ops->fault)
		con->ops->fault(con);
}

S
Sage Weil 已提交
2828 2829 2830
/*
 * Do some work on a connection.  Drop a connection ref when we're done.
 */
2831
static void ceph_con_workfn(struct work_struct *work)
S
Sage Weil 已提交
2832 2833 2834
{
	struct ceph_connection *con = container_of(work, struct ceph_connection,
						   work.work);
2835
	bool fault;
S
Sage Weil 已提交
2836

S
Sage Weil 已提交
2837
	mutex_lock(&con->mutex);
2838 2839
	while (true) {
		int ret;
S
Sage Weil 已提交
2840

2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861
		if ((fault = con_sock_closed(con))) {
			dout("%s: con %p SOCK_CLOSED\n", __func__, con);
			break;
		}
		if (con_backoff(con)) {
			dout("%s: con %p BACKOFF\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_STANDBY) {
			dout("%s: con %p STANDBY\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_CLOSED) {
			dout("%s: con %p CLOSED\n", __func__, con);
			BUG_ON(con->sock);
			break;
		}
		if (con->state == CON_STATE_PREOPEN) {
			dout("%s: con %p PREOPEN\n", __func__, con);
			BUG_ON(con->sock);
		}
2862

2863 2864 2865 2866
		ret = try_read(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2867 2868
			if (!con->error_msg)
				con->error_msg = "socket error on read";
2869 2870 2871 2872 2873 2874 2875 2876
			fault = true;
			break;
		}

		ret = try_write(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2877 2878
			if (!con->error_msg)
				con->error_msg = "socket error on write";
2879 2880 2881 2882
			fault = true;
		}

		break;	/* If we make it to here, we're done */
2883
	}
2884 2885
	if (fault)
		con_fault(con);
S
Sage Weil 已提交
2886
	mutex_unlock(&con->mutex);
2887

2888 2889 2890 2891
	if (fault)
		con_fault_finish(con);

	con->ops->put(con);
S
Sage Weil 已提交
2892 2893 2894 2895 2896 2897
}

/*
 * Generic error/fault handler.  A retry mechanism is used with
 * exponential backoff
 */
2898
static void con_fault(struct ceph_connection *con)
S
Sage Weil 已提交
2899 2900
{
	dout("fault %p state %lu to peer %s\n",
2901
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2902

2903 2904 2905 2906
	pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
		ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
	con->error_msg = NULL;

2907
	WARN_ON(con->state != CON_STATE_CONNECTING &&
2908 2909
	       con->state != CON_STATE_NEGOTIATING &&
	       con->state != CON_STATE_OPEN);
2910

S
Sage Weil 已提交
2911
	con_close_socket(con);
S
Sage Weil 已提交
2912

2913
	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2914 2915
		dout("fault on LOSSYTX channel, marking CLOSED\n");
		con->state = CON_STATE_CLOSED;
2916
		return;
2917 2918
	}

S
Sage Weil 已提交
2919
	if (con->in_msg) {
2920 2921
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
S
Sage Weil 已提交
2922 2923
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
S
Sage Weil 已提交
2924
		con->ops->put(con);
S
Sage Weil 已提交
2925
	}
S
Sage Weil 已提交
2926

2927 2928
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2929

S
Sage Weil 已提交
2930 2931 2932
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2933
	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
2934
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2935
		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2936
		con->state = CON_STATE_STANDBY;
2937 2938
	} else {
		/* retry after a delay. */
2939
		con->state = CON_STATE_PREOPEN;
2940 2941 2942 2943
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
2944
		con_flag_set(con, CON_FLAG_BACKOFF);
A
Alex Elder 已提交
2945
		queue_con(con);
S
Sage Weil 已提交
2946 2947 2948 2949 2950 2951
	}
}



/*
2952
 * initialize a new messenger instance
S
Sage Weil 已提交
2953
 */
2954
void ceph_messenger_init(struct ceph_messenger *msgr,
2955
			 struct ceph_entity_addr *myaddr)
S
Sage Weil 已提交
2956 2957 2958 2959 2960 2961 2962
{
	spin_lock_init(&msgr->global_seq_lock);

	if (myaddr)
		msgr->inst.addr = *myaddr;

	/* select a random nonce */
2963
	msgr->inst.addr.type = 0;
S
Sage Weil 已提交
2964
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2965
	encode_my_addr(msgr);
S
Sage Weil 已提交
2966

2967
	atomic_set(&msgr->stopping, 0);
2968
	write_pnet(&msgr->net, get_net(current->nsproxy->net_ns));
S
Sage Weil 已提交
2969

2970
	dout("%s %p\n", __func__, msgr);
S
Sage Weil 已提交
2971
}
2972
EXPORT_SYMBOL(ceph_messenger_init);
S
Sage Weil 已提交
2973

2974 2975 2976 2977 2978 2979
void ceph_messenger_fini(struct ceph_messenger *msgr)
{
	put_net(read_pnet(&msgr->net));
}
EXPORT_SYMBOL(ceph_messenger_fini);

S
Sage Weil 已提交
2980 2981 2982
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
2983
	if (con->state == CON_STATE_STANDBY) {
S
Sage Weil 已提交
2984
		dout("clear_standby %p and ++connect_seq\n", con);
2985
		con->state = CON_STATE_PREOPEN;
S
Sage Weil 已提交
2986
		con->connect_seq++;
2987 2988
		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
S
Sage Weil 已提交
2989 2990 2991
	}
}

S
Sage Weil 已提交
2992 2993 2994 2995 2996 2997
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	/* set src+dst */
2998
	msg->hdr.src = con->msgr->inst.name;
2999
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
3000 3001
	msg->needs_out_seq = true;

3002
	mutex_lock(&con->mutex);
3003

3004
	if (con->state == CON_STATE_CLOSED) {
3005 3006 3007 3008 3009 3010
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		mutex_unlock(&con->mutex);
		return;
	}

3011
	BUG_ON(msg->con != NULL);
S
Sage Weil 已提交
3012
	msg->con = con->ops->get(con);
3013 3014
	BUG_ON(msg->con == NULL);

S
Sage Weil 已提交
3015 3016 3017 3018 3019 3020 3021 3022
	BUG_ON(!list_empty(&msg->list_head));
	list_add_tail(&msg->list_head, &con->out_queue);
	dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
	     ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
	     le32_to_cpu(msg->hdr.front_len),
	     le32_to_cpu(msg->hdr.middle_len),
	     le32_to_cpu(msg->hdr.data_len));
3023 3024

	clear_standby(con);
3025
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3026 3027 3028

	/* if there wasn't anything waiting to send before, queue
	 * new work */
3029
	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
3030 3031
		queue_con(con);
}
3032
EXPORT_SYMBOL(ceph_con_send);
S
Sage Weil 已提交
3033 3034 3035 3036

/*
 * Revoke a message that was previously queued for send
 */
3037
void ceph_msg_revoke(struct ceph_msg *msg)
S
Sage Weil 已提交
3038
{
3039 3040 3041 3042 3043
	struct ceph_connection *con = msg->con;

	if (!con)
		return;		/* Message not in our possession */

3044
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3045
	if (!list_empty(&msg->list_head)) {
3046
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
S
Sage Weil 已提交
3047
		list_del_init(&msg->list_head);
3048
		BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
3049
		msg->con->ops->put(msg->con);
3050
		msg->con = NULL;
S
Sage Weil 已提交
3051
		msg->hdr.seq = 0;
3052

S
Sage Weil 已提交
3053
		ceph_msg_put(msg);
S
Sage Weil 已提交
3054 3055
	}
	if (con->out_msg == msg) {
3056
		dout("%s %p msg %p - was sending\n", __func__, con, msg);
S
Sage Weil 已提交
3057
		con->out_msg = NULL;
S
Sage Weil 已提交
3058 3059 3060 3061
		if (con->out_kvec_is_msg) {
			con->out_skip = con->out_kvec_bytes;
			con->out_kvec_is_msg = false;
		}
S
Sage Weil 已提交
3062
		msg->hdr.seq = 0;
3063 3064

		ceph_msg_put(msg);
S
Sage Weil 已提交
3065
	}
3066
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3067 3068
}

3069
/*
3070
 * Revoke a message that we may be reading data into
3071
 */
3072
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
3073
{
3074 3075 3076 3077 3078 3079 3080 3081 3082 3083
	struct ceph_connection *con;

	BUG_ON(msg == NULL);
	if (!msg->con) {
		dout("%s msg %p null con\n", __func__, msg);

		return;		/* Message not in our possession */
	}

	con = msg->con;
3084
	mutex_lock(&con->mutex);
3085
	if (con->in_msg == msg) {
3086 3087 3088
		unsigned int front_len = le32_to_cpu(con->in_hdr.front_len);
		unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len);
		unsigned int data_len = le32_to_cpu(con->in_hdr.data_len);
3089 3090

		/* skip rest of message */
3091 3092
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
3093
				sizeof(struct ceph_msg_header) -
3094 3095 3096
				front_len -
				middle_len -
				data_len -
3097 3098 3099 3100
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
3101
		con->in_seq++;
3102
	} else {
3103 3104
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
3105 3106 3107 3108
	}
	mutex_unlock(&con->mutex);
}

S
Sage Weil 已提交
3109 3110 3111 3112 3113
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
S
Sage Weil 已提交
3114
	dout("con_keepalive %p\n", con);
3115
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3116
	clear_standby(con);
3117
	mutex_unlock(&con->mutex);
3118 3119
	if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 &&
	    con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
3120 3121
		queue_con(con);
}
3122
EXPORT_SYMBOL(ceph_con_keepalive);
S
Sage Weil 已提交
3123

3124 3125 3126 3127 3128 3129 3130 3131 3132 3133 3134 3135 3136 3137
bool ceph_con_keepalive_expired(struct ceph_connection *con,
			       unsigned long interval)
{
	if (interval > 0 &&
	    (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2)) {
		struct timespec now = CURRENT_TIME;
		struct timespec ts;
		jiffies_to_timespec(interval, &ts);
		ts = timespec_add(con->last_keepalive_ack, ts);
		return timespec_compare(&now, &ts) >= 0;
	}
	return false;
}

3138
static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type)
A
Alex Elder 已提交
3139
{
3140 3141 3142 3143 3144
	struct ceph_msg_data *data;

	if (WARN_ON(!ceph_msg_data_type_valid(type)))
		return NULL;

3145
	data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS);
3146 3147
	if (data)
		data->type = type;
3148
	INIT_LIST_HEAD(&data->links);
3149 3150 3151 3152 3153 3154 3155 3156 3157

	return data;
}

static void ceph_msg_data_destroy(struct ceph_msg_data *data)
{
	if (!data)
		return;

3158
	WARN_ON(!list_empty(&data->links));
Y
Yan, Zheng 已提交
3159
	if (data->type == CEPH_MSG_DATA_PAGELIST)
3160
		ceph_pagelist_release(data->pagelist);
3161
	kmem_cache_free(ceph_msg_data_cache, data);
A
Alex Elder 已提交
3162 3163
}

3164
void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
3165
		size_t length, size_t alignment)
3166
{
3167 3168
	struct ceph_msg_data *data;

3169 3170
	BUG_ON(!pages);
	BUG_ON(!length);
3171 3172 3173 3174 3175 3176

	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
	BUG_ON(!data);
	data->pages = pages;
	data->length = length;
	data->alignment = alignment & ~PAGE_MASK;
3177

3178 3179
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3180
}
3181
EXPORT_SYMBOL(ceph_msg_data_add_pages);
S
Sage Weil 已提交
3182

3183
void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
3184 3185
				struct ceph_pagelist *pagelist)
{
3186 3187
	struct ceph_msg_data *data;

3188 3189
	BUG_ON(!pagelist);
	BUG_ON(!pagelist->length);
3190

3191 3192 3193 3194
	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
	BUG_ON(!data);
	data->pagelist = pagelist;

3195 3196
	list_add_tail(&data->links, &msg->data);
	msg->data_length += pagelist->length;
3197
}
3198
EXPORT_SYMBOL(ceph_msg_data_add_pagelist);
3199

3200
#ifdef	CONFIG_BLOCK
3201
void ceph_msg_data_add_bio(struct ceph_msg *msg, struct bio *bio,
A
Alex Elder 已提交
3202
		size_t length)
3203
{
3204 3205
	struct ceph_msg_data *data;

3206
	BUG_ON(!bio);
3207

3208 3209 3210
	data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
	BUG_ON(!data);
	data->bio = bio;
A
Alex Elder 已提交
3211
	data->bio_length = length;
3212

3213 3214
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3215
}
3216
EXPORT_SYMBOL(ceph_msg_data_add_bio);
3217
#endif	/* CONFIG_BLOCK */
3218

S
Sage Weil 已提交
3219 3220 3221 3222
/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
3223 3224
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
S
Sage Weil 已提交
3225 3226 3227
{
	struct ceph_msg *m;

3228
	m = kmem_cache_zalloc(ceph_msg_cache, flags);
S
Sage Weil 已提交
3229 3230 3231 3232
	if (m == NULL)
		goto out;

	m->hdr.type = cpu_to_le16(type);
3233
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
S
Sage Weil 已提交
3234
	m->hdr.front_len = cpu_to_le32(front_len);
3235

3236 3237
	INIT_LIST_HEAD(&m->list_head);
	kref_init(&m->kref);
3238
	INIT_LIST_HEAD(&m->data);
3239

S
Sage Weil 已提交
3240 3241
	/* front */
	if (front_len) {
3242
		m->front.iov_base = ceph_kvmalloc(front_len, flags);
S
Sage Weil 已提交
3243
		if (m->front.iov_base == NULL) {
3244
			dout("ceph_msg_new can't allocate %d bytes\n",
S
Sage Weil 已提交
3245 3246 3247 3248 3249 3250
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
3251
	m->front_alloc_len = m->front.iov_len = front_len;
S
Sage Weil 已提交
3252

S
Sage Weil 已提交
3253
	dout("ceph_msg_new %p front %d\n", m, front_len);
S
Sage Weil 已提交
3254 3255 3256 3257 3258
	return m;

out2:
	ceph_msg_put(m);
out:
3259 3260 3261
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
3262
		WARN_ON(1);
3263 3264 3265 3266
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
3267
	return NULL;
S
Sage Weil 已提交
3268
}
3269
EXPORT_SYMBOL(ceph_msg_new);
S
Sage Weil 已提交
3270 3271 3272 3273 3274 3275 3276 3277

/*
 * Allocate "middle" portion of a message, if it is needed and wasn't
 * allocated by alloc_msg.  This allows us to read a small fixed-size
 * per-type header in the front and then gracefully fail (i.e.,
 * propagate the error to the caller based on info in the front) when
 * the middle is too large.
 */
3278
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
S
Sage Weil 已提交
3279 3280 3281 3282 3283 3284 3285 3286 3287
{
	int type = le16_to_cpu(msg->hdr.type);
	int middle_len = le32_to_cpu(msg->hdr.middle_len);

	dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
	     ceph_msg_type_name(type), middle_len);
	BUG_ON(!middle_len);
	BUG_ON(msg->middle);

S
Sage Weil 已提交
3288
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
S
Sage Weil 已提交
3289 3290 3291 3292 3293
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

3294
/*
A
Alex Elder 已提交
3295 3296 3297 3298
 * Allocate a message for receiving an incoming message on a
 * connection, and save the result in con->in_msg.  Uses the
 * connection's private alloc_msg op if available.
 *
3299 3300 3301 3302 3303 3304 3305 3306 3307
 * Returns 0 on success, or a negative error code.
 *
 * On success, if we set *skip = 1:
 *  - the next message should be skipped and ignored.
 *  - con->in_msg == NULL
 * or if we set *skip = 0:
 *  - con->in_msg is non-null.
 * On error (ENOMEM, EAGAIN, ...),
 *  - con->in_msg == NULL
3308
 */
3309
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3310
{
3311
	struct ceph_msg_header *hdr = &con->in_hdr;
3312
	int middle_len = le32_to_cpu(hdr->middle_len);
3313
	struct ceph_msg *msg;
3314
	int ret = 0;
3315

A
Alex Elder 已提交
3316
	BUG_ON(con->in_msg != NULL);
3317
	BUG_ON(!con->ops->alloc_msg);
3318

3319 3320 3321 3322 3323
	mutex_unlock(&con->mutex);
	msg = con->ops->alloc_msg(con, hdr, skip);
	mutex_lock(&con->mutex);
	if (con->state != CON_STATE_OPEN) {
		if (msg)
3324
			ceph_msg_put(msg);
3325 3326
		return -EAGAIN;
	}
3327 3328 3329
	if (msg) {
		BUG_ON(*skip);
		con->in_msg = msg;
S
Sage Weil 已提交
3330
		con->in_msg->con = con->ops->get(con);
3331
		BUG_ON(con->in_msg->con == NULL);
3332 3333 3334 3335 3336 3337 3338 3339 3340
	} else {
		/*
		 * Null message pointer means either we should skip
		 * this message or we couldn't allocate memory.  The
		 * former is not an error.
		 */
		if (*skip)
			return 0;

3341
		con->error_msg = "error allocating memory for incoming message";
3342
		return -ENOMEM;
3343
	}
A
Alex Elder 已提交
3344
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3345

A
Alex Elder 已提交
3346 3347
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
3348
		if (ret < 0) {
A
Alex Elder 已提交
3349 3350
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
3351 3352
		}
	}
3353

3354
	return ret;
3355 3356
}

S
Sage Weil 已提交
3357 3358 3359 3360

/*
 * Free a generically kmalloc'd message.
 */
3361
static void ceph_msg_free(struct ceph_msg *m)
S
Sage Weil 已提交
3362
{
3363
	dout("%s %p\n", __func__, m);
I
Ilya Dryomov 已提交
3364
	kvfree(m->front.iov_base);
3365
	kmem_cache_free(ceph_msg_cache, m);
S
Sage Weil 已提交
3366 3367
}

3368
static void ceph_msg_release(struct kref *kref)
S
Sage Weil 已提交
3369 3370
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
3371 3372 3373
	LIST_HEAD(data);
	struct list_head *links;
	struct list_head *next;
S
Sage Weil 已提交
3374

3375
	dout("%s %p\n", __func__, m);
S
Sage Weil 已提交
3376 3377 3378 3379 3380 3381
	WARN_ON(!list_empty(&m->list_head));

	/* drop middle, data, if any */
	if (m->middle) {
		ceph_buffer_put(m->middle);
		m->middle = NULL;
S
Sage Weil 已提交
3382
	}
3383 3384 3385 3386 3387 3388 3389 3390 3391

	list_splice_init(&m->data, &data);
	list_for_each_safe(links, next, &data) {
		struct ceph_msg_data *data;

		data = list_entry(links, struct ceph_msg_data, links);
		list_del_init(links);
		ceph_msg_data_destroy(data);
	}
A
Alex Elder 已提交
3392
	m->data_length = 0;
3393

S
Sage Weil 已提交
3394 3395 3396
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
3397 3398 3399 3400 3401 3402 3403 3404 3405 3406 3407 3408 3409 3410 3411 3412 3413
		ceph_msg_free(m);
}

struct ceph_msg *ceph_msg_get(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
	     atomic_read(&msg->kref.refcount));
	kref_get(&msg->kref);
	return msg;
}
EXPORT_SYMBOL(ceph_msg_get);

void ceph_msg_put(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
	     atomic_read(&msg->kref.refcount));
	kref_put(&msg->kref, ceph_msg_release);
S
Sage Weil 已提交
3414
}
3415
EXPORT_SYMBOL(ceph_msg_put);
3416 3417 3418

void ceph_msg_dump(struct ceph_msg *msg)
{
3419 3420
	pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg,
		 msg->front_alloc_len, msg->data_length);
3421 3422 3423 3424 3425 3426 3427 3428 3429 3430 3431 3432 3433 3434 3435
	print_hex_dump(KERN_DEBUG, "header: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       &msg->hdr, sizeof(msg->hdr), true);
	print_hex_dump(KERN_DEBUG, " front: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       msg->front.iov_base, msg->front.iov_len, true);
	if (msg->middle)
		print_hex_dump(KERN_DEBUG, "middle: ",
			       DUMP_PREFIX_OFFSET, 16, 1,
			       msg->middle->vec.iov_base,
			       msg->middle->vec.iov_len, true);
	print_hex_dump(KERN_DEBUG, "footer: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       &msg->footer, sizeof(msg->footer), true);
}
3436
EXPORT_SYMBOL(ceph_msg_dump);