messenger.c 83.7 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/slab.h>
S
Sage Weil 已提交
10 11
#include <linux/socket.h>
#include <linux/string.h>
12
#ifdef	CONFIG_BLOCK
13
#include <linux/bio.h>
14
#endif	/* CONFIG_BLOCK */
N
Noah Watkins 已提交
15
#include <linux/dns_resolver.h>
S
Sage Weil 已提交
16 17
#include <net/tcp.h>

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

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

S
Sage Weil 已提交
28 29 30 31 32 33 34 35 36
/*
 * 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 已提交
37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55
/*
 * 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()       \
56 57 58 59 60 61 62 63 64 65 66
 *       |+---------------------------    \
 *       | \                          \    \
 *       |  -----------                \    \
 *       |  | CLOSING |  socket event;  \    \
 *       |  -----------  await close     \    \
 *       |       ^                        \   |
 *       |       |                         \  |
 *       |       + con_sock_state_closing() \ |
 *       |      / \                         | |
 *       |     /   ---------------          | |
 *       |    /                   \         v v
A
Alex Elder 已提交
67 68 69 70 71 72 73 74 75 76 77
 *       |   /                    --------------
 *       |  /    -----------------| 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.
 */
78 79 80 81 82 83 84

#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 */

85 86 87 88 89 90 91 92 93 94
/*
 * 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 已提交
95 96 97 98 99 100 101 102 103
/*
 * 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 */
104

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
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);
}

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

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

S
Sage Weil 已提交
161 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 167 168 169
#ifdef CONFIG_LOCKDEP
static struct lock_class_key socket_class;
#endif

A
Alex Elder 已提交
170 171 172 173 174
/*
 * 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 已提交
175 176

static void queue_con(struct ceph_connection *con);
177
static void cancel_con(struct ceph_connection *con);
S
Sage Weil 已提交
178
static void con_work(struct work_struct *);
179
static void con_fault(struct ceph_connection *con);
S
Sage Weil 已提交
180 181

/*
182 183
 * Nicely render a sockaddr as a string.  An array of formatted
 * strings is used, to approximate reentrancy.
S
Sage Weil 已提交
184
 */
185 186 187 188 189 190 191
#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 已提交
192

193 194
static struct page *zero_page;		/* used in certain error cases */

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

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

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

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

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

	return s;
}
223
EXPORT_SYMBOL(ceph_pr_addr);
S
Sage Weil 已提交
224

225 226 227 228 229 230
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 已提交
231 232 233
/*
 * work queue for all reading and writing to/from the socket.
 */
A
Alex Elder 已提交
234
static struct workqueue_struct *ceph_msgr_wq;
S
Sage Weil 已提交
235

236 237 238 239 240 241
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);
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257

	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;
258 259 260 261
}

static void ceph_msgr_slab_exit(void)
{
262 263 264 265
	BUG_ON(!ceph_msg_data_cache);
	kmem_cache_destroy(ceph_msg_data_cache);
	ceph_msg_data_cache = NULL;

266 267 268 269 270
	BUG_ON(!ceph_msg_cache);
	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;
}

A
Alex Elder 已提交
271
static void _ceph_msgr_exit(void)
272
{
A
Alex Elder 已提交
273
	if (ceph_msgr_wq) {
274
		destroy_workqueue(ceph_msgr_wq);
A
Alex Elder 已提交
275 276
		ceph_msgr_wq = NULL;
	}
277

278 279
	ceph_msgr_slab_exit();

280 281 282 283 284 285
	BUG_ON(zero_page == NULL);
	kunmap(zero_page);
	page_cache_release(zero_page);
	zero_page = NULL;
}

286
int ceph_msgr_init(void)
S
Sage Weil 已提交
287
{
288 289 290 291
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
	page_cache_get(zero_page);

292 293 294
	if (ceph_msgr_slab_init())
		return -ENOMEM;

295 296 297 298 299
	/*
	 * 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);
300 301
	if (ceph_msgr_wq)
		return 0;
302

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

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

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

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

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

324 325 326 327 328 329 330 331 332
/* 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);
333 334
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
335 336 337 338 339 340 341 342 343
}

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);
344 345
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTING);
346 347 348 349 350 351 352 353 354
}

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);
355 356
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTED);
357 358 359 360 361 362 363 364 365 366 367
}

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);
368 369
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSING);
370 371 372 373 374 375 376 377
}

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 &&
378
		    old_state != CON_SOCK_STATE_CLOSING &&
379 380
		    old_state != CON_SOCK_STATE_CONNECTING &&
		    old_state != CON_SOCK_STATE_CLOSED))
381
		printk("%s: unexpected old state %d\n", __func__, old_state);
382 383
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
384
}
385

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

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

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

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

410 411
	/* 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 已提交
412
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
413 414 415 416
	 * 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().
	 */
417
	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
418
		if (sk_stream_is_writeable(sk)) {
A
Alex Elder 已提交
419
			dout("%s %p queueing write work\n", __func__, con);
420 421 422
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
S
Sage Weil 已提交
423
	} else {
A
Alex Elder 已提交
424
		dout("%s %p nothing to write\n", __func__, con);
S
Sage Weil 已提交
425 426 427 428
	}
}

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

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

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

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
A
Alex Elder 已提交
462
	sk->sk_user_data = con;
A
Alex Elder 已提交
463 464 465
	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 已提交
466 467 468 469 470 471 472 473 474 475
}


/*
 * socket helpers
 */

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

	BUG_ON(con->sock);
S
Sage Weil 已提交
483 484
	ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
			       IPPROTO_TCP, &sock);
S
Sage Weil 已提交
485
	if (ret)
486
		return ret;
487
	sock->sk->sk_allocation = GFP_NOFS | __GFP_MEMALLOC;
S
Sage Weil 已提交
488

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

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

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

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

510
		return ret;
511
	}
512

C
Chaitanya Huilgol 已提交
513 514 515 516 517 518 519 520 521 522
	if (con->msgr->tcp_nodelay) {
		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);
	}

523 524
	sk_set_memalloc(sock->sk);

525
	con->sock = sock;
526
	return 0;
S
Sage Weil 已提交
527 528 529 530 531 532
}

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 };
533
	int r;
S
Sage Weil 已提交
534

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

541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556
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 已提交
557 558 559 560 561 562 563 564
/*
 * 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 };
565
	int r;
S
Sage Weil 已提交
566 567 568 569 570 571

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

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

578
static int __ceph_tcp_sendpage(struct socket *sock, struct page *page,
579
		     int offset, size_t size, bool more)
580 581 582 583 584 585 586 587 588 589 590
{
	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;
}

591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608
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 已提交
609 610 611 612 613 614

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

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

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

632
	con_sock_state_closed(con);
S
Sage Weil 已提交
633 634 635 636 637 638 639 640 641 642
	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);
643
	BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
644
	msg->con->ops->put(msg->con);
645 646
	msg->con = NULL;

S
Sage Weil 已提交
647 648 649 650 651 652 653 654 655 656 657 658 659 660 661
	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 */
662
	dout("reset_connection %p\n", con);
S
Sage Weil 已提交
663 664 665
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

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

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

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

694 695 696 697
	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);
698

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

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

717
	WARN_ON(con->state != CON_STATE_CLOSED);
718
	con->state = CON_STATE_PREOPEN;
719

720 721 722
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

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

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

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

	con_sock_state_init(con);

753
	mutex_init(&con->mutex);
S
Sage Weil 已提交
754 755 756
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
	INIT_DELAYED_WORK(&con->work, con_work);
757

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


/*
 * 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;
}

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

786
static void con_out_kvec_add(struct ceph_connection *con,
787 788 789 790 791 792 793 794 795 796 797 798
				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 已提交
799

800
#ifdef CONFIG_BLOCK
801 802 803 804 805 806

/*
 * 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 已提交
807
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
808
					size_t length)
809
{
A
Alex Elder 已提交
810
	struct ceph_msg_data *data = cursor->data;
811 812 813 814 815 816 817
	struct bio *bio;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

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

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

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

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

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

838
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
839

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

849
	return bio_vec.bv_page;
850 851
}

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

A
Alex Elder 已提交
858
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_BIO);
859 860 861 862

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

863
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
864 865 866

	/* Advance the cursor offset */

867 868
	BUG_ON(cursor->resid < bytes);
	cursor->resid -= bytes;
869 870 871 872

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

	if (bytes < bio_vec.bv_len)
873 874 875 876
		return false;	/* more bytes to process in this segment */

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

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

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

	return true;
}
897
#endif /* CONFIG_BLOCK */
898

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

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

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

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

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

	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;
936
	if (cursor->last_piece)
937
		*length = cursor->resid;
938
	else
939 940 941 942 943
		*length = PAGE_SIZE - *page_offset;

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

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

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

	/* Advance the cursor page offset */

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

958 959 960
	if (!cursor->resid)
		return false;   /* no more data */

961
	/* Move on to the next page; offset is already at 0 */
962 963 964

	BUG_ON(cursor->page_index >= cursor->page_count);
	cursor->page_index++;
965
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
966 967 968 969

	return true;
}

970
/*
971 972
 * 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.
973
 */
A
Alex Elder 已提交
974 975
static void
ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
976
					size_t length)
977
{
A
Alex Elder 已提交
978
	struct ceph_msg_data *data = cursor->data;
979 980 981
	struct ceph_pagelist *pagelist;
	struct page *page;

982
	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
983 984 985

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

	if (!length)
988 989 990 991 992
		return;		/* pagelist can be assigned but empty */

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

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

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

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(!cursor->page);
1012
	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1013

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

A
Alex Elder 已提交
1021
	return cursor->page;
1022 1023
}

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

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1036 1037 1038 1039
	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);

	/* Advance the cursor offset */

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

1046 1047 1048
	if (!cursor->resid)
		return false;   /* no more data */

1049 1050 1051 1052
	/* Move on to the next page */

	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
	cursor->page = list_entry_next(cursor->page, lru);
1053
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1054 1055 1056 1057

	return true;
}

1058 1059 1060 1061 1062 1063 1064 1065
/*
 * 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.
 */
1066
static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
1067
{
1068
	size_t length = cursor->total_resid;
A
Alex Elder 已提交
1069 1070

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

1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106
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);
}

1107 1108 1109 1110 1111
/*
 * 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 已提交
1112 1113
static struct page *ceph_msg_data_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length,
1114 1115 1116 1117
					bool *last_piece)
{
	struct page *page;

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

	return page;
}

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

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

1173 1174 1175 1176 1177
	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 已提交
1178
		new_piece = true;
1179
	}
A
Alex Elder 已提交
1180
	cursor->need_crc = new_piece;
1181

1182 1183 1184
	return new_piece;
}

1185
static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
1186 1187
{
	BUG_ON(!msg);
1188
	BUG_ON(!data_len);
1189

1190
	/* Initialize data cursor */
1191

A
Alex Elder 已提交
1192
	ceph_msg_data_cursor_init(msg, (size_t)data_len);
1193 1194
}

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

1204 1205
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

S
Sage Weil 已提交
1206 1207 1208
	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 已提交
1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220
	if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
		if (con->ops->sign_message)
			con->ops->sign_message(con, m);
		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 已提交
1221 1222
	con->out_kvec_left++;
	con->out_more = m->more_to_follow;
1223
	con->out_msg_done = true;
S
Sage Weil 已提交
1224 1225 1226 1227 1228 1229 1230 1231
}

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

1234
	con_out_kvec_reset(con);
S
Sage Weil 已提交
1235
	con->out_kvec_is_msg = true;
1236
	con->out_msg_done = false;
S
Sage Weil 已提交
1237 1238 1239 1240 1241

	/* 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;
1242
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
S
Sage Weil 已提交
1243
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1244
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1245
			&con->out_temp_ack);
S
Sage Weil 已提交
1246 1247
	}

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

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

1257 1258 1259 1260 1261 1262 1263 1264
	/*
	 * 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;
	}
1265
	WARN_ON(m->data_length != le32_to_cpu(m->hdr.data_len));
S
Sage Weil 已提交
1266

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

	/* tag + hdr + front + middle */
1274 1275 1276
	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);
1277

S
Sage Weil 已提交
1278
	if (m->middle)
1279
		con_out_kvec_add(con, m->middle->vec.iov_len,
1280
			m->middle->vec.iov_base);
S
Sage Weil 已提交
1281 1282

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

	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 已提交
1294
		con->out_msg->footer.middle_crc = 0;
1295
	dout("%s front_crc %u middle_crc %u\n", __func__,
S
Sage Weil 已提交
1296 1297 1298 1299
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));

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

1309
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320
}

/*
 * 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;

1321
	con_out_kvec_reset(con);
1322

1323
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1324

S
Sage Weil 已提交
1325
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1326
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1327 1328
				&con->out_temp_ack);

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

1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350
/*
 * 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 已提交
1351 1352 1353 1354 1355 1356
/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
1357 1358
	con_out_kvec_reset(con);
	con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
1359
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1360 1361 1362 1363 1364 1365
}

/*
 * Connection negotiation.
 */

1366 1367
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
1368
{
1369
	struct ceph_auth_handshake *auth;
1370 1371 1372 1373

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;
1374
		return NULL;
1375 1376 1377
	}

	/* Can't hold the mutex while getting authorizer */
1378
	mutex_unlock(&con->mutex);
1379
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1380
	mutex_lock(&con->mutex);
1381

1382
	if (IS_ERR(auth))
1383
		return auth;
1384
	if (con->state != CON_STATE_NEGOTIATING)
1385
		return ERR_PTR(-EAGAIN);
1386

1387 1388
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1389
	return auth;
1390 1391
}

S
Sage Weil 已提交
1392 1393 1394
/*
 * We connected to a peer and are saying hello.
 */
1395
static void prepare_write_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1396
{
1397 1398
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1399
					&con->msgr->my_enc_addr);
1400 1401

	con->out_more = 0;
1402
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1403 1404
}

1405
static int prepare_write_connect(struct ceph_connection *con)
1406
{
1407
	unsigned int global_seq = get_global_seq(con->msgr, 0);
S
Sage Weil 已提交
1408
	int proto;
1409
	int auth_proto;
1410
	struct ceph_auth_handshake *auth;
S
Sage Weil 已提交
1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427

	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);
1428

1429
	con->out_connect.features = cpu_to_le64(con->msgr->supported_features);
S
Sage Weil 已提交
1430 1431 1432 1433 1434 1435
	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;

1436 1437
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
1438 1439
	if (IS_ERR(auth))
		return PTR_ERR(auth);
1440

1441
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1442 1443 1444
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

1445
	con_out_kvec_add(con, sizeof (con->out_connect),
1446 1447
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
1448
		con_out_kvec_add(con, auth->authorizer_buf_len,
1449
					auth->authorizer_buf);
1450

S
Sage Weil 已提交
1451
	con->out_more = 0;
1452
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1453

1454
	return 0;
S
Sage Weil 已提交
1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476
}

/*
 * 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 */
1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488

		/* 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 已提交
1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499
		}
	}
	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! */
}

1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512
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 已提交
1513 1514 1515 1516 1517 1518 1519
/*
 * 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
 */
1520
static int write_partial_message_data(struct ceph_connection *con)
S
Sage Weil 已提交
1521 1522
{
	struct ceph_msg *msg = con->out_msg;
A
Alex Elder 已提交
1523
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1524
	bool do_datacrc = !con->msgr->nocrc;
A
Alex Elder 已提交
1525
	u32 crc;
S
Sage Weil 已提交
1526

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

1529
	if (list_empty(&msg->data))
1530 1531
		return -EINVAL;

1532 1533 1534 1535 1536 1537 1538 1539
	/*
	 * 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 已提交
1540
	crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0;
1541
	while (cursor->resid) {
1542
		struct page *page;
1543 1544
		size_t page_offset;
		size_t length;
1545
		bool last_piece;
1546
		bool need_crc;
A
Alex Elder 已提交
1547
		int ret;
1548

A
Alex Elder 已提交
1549
		page = ceph_msg_data_next(&msg->cursor, &page_offset, &length,
1550
							&last_piece);
1551
		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1552
				      length, last_piece);
A
Alex Elder 已提交
1553 1554 1555
		if (ret <= 0) {
			if (do_datacrc)
				msg->footer.data_crc = cpu_to_le32(crc);
S
Sage Weil 已提交
1556

A
Alex Elder 已提交
1557 1558
			return ret;
		}
1559 1560
		if (do_datacrc && cursor->need_crc)
			crc = ceph_crc32c_page(crc, page, page_offset, length);
A
Alex Elder 已提交
1561
		need_crc = ceph_msg_data_advance(&msg->cursor, (size_t)ret);
S
Sage Weil 已提交
1562 1563
	}

1564
	dout("%s %p msg %p done\n", __func__, con, msg);
S
Sage Weil 已提交
1565 1566

	/* prepare and queue up footer, too */
A
Alex Elder 已提交
1567 1568 1569
	if (do_datacrc)
		msg->footer.data_crc = cpu_to_le32(crc);
	else
1570
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1571
	con_out_kvec_reset(con);
1572
	prepare_write_message_footer(con);
A
Alex Elder 已提交
1573 1574

	return 1;	/* must return > 0 to indicate success */
S
Sage Weil 已提交
1575 1576 1577 1578 1579 1580 1581 1582 1583 1584
}

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

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

1587
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
S
Sage Weil 已提交
1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1600 1601 1602 1603 1604 1605
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617
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;
}

1618 1619 1620 1621 1622 1623 1624
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 已提交
1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645
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;
}

/*
 * 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,
1646
			int end, int size, void *object)
S
Sage Weil 已提交
1647
{
1648 1649
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
S
Sage Weil 已提交
1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662
		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
 */
1663
static int read_partial_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1664
{
1665 1666 1667
	int size;
	int end;
	int ret;
S
Sage Weil 已提交
1668

1669
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
S
Sage Weil 已提交
1670 1671

	/* peer's banner */
1672 1673 1674
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
S
Sage Weil 已提交
1675 1676
	if (ret <= 0)
		goto out;
1677 1678 1679 1680

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
S
Sage Weil 已提交
1681 1682
	if (ret <= 0)
		goto out;
1683 1684 1685 1686

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

1690 1691 1692 1693 1694 1695
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1696 1697 1698
	int size;
	int end;
	int ret;
1699 1700 1701

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

1702 1703 1704
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
S
Sage Weil 已提交
1705 1706
	if (ret <= 0)
		goto out;
1707 1708 1709 1710

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

1714 1715 1716
	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 已提交
1717 1718 1719
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1720

S
Sage Weil 已提交
1721 1722 1723 1724 1725 1726 1727 1728
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1729
		pr_err("connect to %s got bad banner\n",
1730
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755
		con->error_msg = "protocol error, bad banner";
		return -1;
	}
	return 0;
}

static bool addr_is_blank(struct sockaddr_storage *ss)
{
	switch (ss->ss_family) {
	case AF_INET:
		return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
	case AF_INET6:
		return
		     ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
		     ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
		     ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
		     ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
	}
	return false;
}

static int addr_port(struct sockaddr_storage *ss)
{
	switch (ss->ss_family) {
	case AF_INET:
S
Sage Weil 已提交
1756
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
S
Sage Weil 已提交
1757
	case AF_INET6:
S
Sage Weil 已提交
1758
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
S
Sage Weil 已提交
1759 1760 1761 1762 1763 1764 1765 1766 1767
	}
	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);
1768
		break;
S
Sage Weil 已提交
1769 1770
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1771
		break;
S
Sage Weil 已提交
1772 1773 1774
	}
}

N
Noah Watkins 已提交
1775 1776 1777 1778 1779 1780
/*
 * 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 已提交
1781 1782
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
N
Noah Watkins 已提交
1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 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

	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 已提交
1870 1871 1872 1873 1874 1875 1876 1877
/*
 * 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 已提交
1878
	int i, ret = -EINVAL;
S
Sage Weil 已提交
1879 1880 1881 1882 1883 1884 1885
	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 已提交
1886 1887 1888 1889 1890 1891
		char delim = ',';

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

N
Noah Watkins 已提交
1893 1894
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
S
Sage Weil 已提交
1895
			goto bad;
N
Noah Watkins 已提交
1896 1897
		ret = -EINVAL;

S
Sage Weil 已提交
1898 1899
		p = ipend;

S
Sage Weil 已提交
1900 1901 1902 1903 1904 1905 1906 1907
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

S
Sage Weil 已提交
1908 1909 1910 1911 1912 1913 1914 1915
		/* port? */
		if (p < end && *p == ':') {
			port = 0;
			p++;
			while (p < end && *p >= '0' && *p <= '9') {
				port = (port * 10) + (*p - '0');
				p++;
			}
1916 1917 1918
			if (port == 0)
				port = CEPH_MON_PORT;
			else if (port > 65535)
S
Sage Weil 已提交
1919 1920 1921 1922 1923 1924 1925
				goto bad;
		} else {
			port = CEPH_MON_PORT;
		}

		addr_set_port(ss, port);

1926
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
S
Sage Weil 已提交
1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942

		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 已提交
1943
	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
N
Noah Watkins 已提交
1944
	return ret;
S
Sage Weil 已提交
1945
}
1946
EXPORT_SYMBOL(ceph_parse_ips);
S
Sage Weil 已提交
1947

1948
static int process_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1949
{
1950
	dout("process_banner on %p\n", con);
S
Sage Weil 已提交
1951 1952 1953 1954

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

1955 1956 1957
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

S
Sage Weil 已提交
1958 1959 1960 1961 1962
	/*
	 * 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 已提交
1963 1964
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
S
Sage Weil 已提交
1965 1966
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
1967 1968 1969 1970 1971
		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));
1972
		con->error_msg = "wrong peer at address";
S
Sage Weil 已提交
1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985
		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);
1986
		encode_my_addr(con->msgr);
1987
		dout("process_banner learned my addr is %s\n",
1988
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
S
Sage Weil 已提交
1989 1990
	}

1991 1992 1993 1994 1995
	return 0;
}

static int process_connect(struct ceph_connection *con)
{
1996 1997
	u64 sup_feat = con->msgr->supported_features;
	u64 req_feat = con->msgr->required_features;
I
Ilya Dryomov 已提交
1998 1999
	u64 server_feat = ceph_sanitize_features(
				le64_to_cpu(con->in_reply.features));
2000
	int ret;
2001

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

S
Sage Weil 已提交
2004
	switch (con->in_reply.tag) {
2005 2006 2007 2008
	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),
2009
		       ceph_pr_addr(&con->peer_addr.in_addr),
2010 2011
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
2012
		reset_connection(con);
2013 2014
		return -1;

S
Sage Weil 已提交
2015 2016 2017 2018
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
2019
		       ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
2020 2021 2022
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
2023
		reset_connection(con);
S
Sage Weil 已提交
2024 2025
		return -1;

2026 2027 2028 2029 2030 2031 2032 2033
	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;
		}
2034
		con_out_kvec_reset(con);
2035
		ret = prepare_write_connect(con);
2036 2037
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2038
		prepare_read_connect(con);
2039
		break;
S
Sage Weil 已提交
2040 2041 2042 2043 2044 2045 2046 2047 2048 2049

	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 已提交
2050
		     le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2051 2052
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
2053
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2054
		reset_connection(con);
2055
		con_out_kvec_reset(con);
2056 2057 2058
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2059 2060 2061
		prepare_read_connect(con);

		/* Tell ceph about it. */
2062
		mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2063 2064 2065
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
2066
		mutex_lock(&con->mutex);
2067
		if (con->state != CON_STATE_NEGOTIATING)
2068
			return -EAGAIN;
S
Sage Weil 已提交
2069 2070 2071 2072 2073 2074 2075
		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 已提交
2076
		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
S
Sage Weil 已提交
2077
		     le32_to_cpu(con->out_connect.connect_seq),
S
Sage Weil 已提交
2078 2079
		     le32_to_cpu(con->in_reply.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2080
		con_out_kvec_reset(con);
2081 2082 2083
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2084 2085 2086 2087 2088 2089 2090 2091
		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.
		 */
2092
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
S
Sage Weil 已提交
2093
		     con->peer_global_seq,
S
Sage Weil 已提交
2094
		     le32_to_cpu(con->in_reply.global_seq));
S
Sage Weil 已提交
2095
		get_global_seq(con->msgr,
S
Sage Weil 已提交
2096
			       le32_to_cpu(con->in_reply.global_seq));
2097
		con_out_kvec_reset(con);
2098 2099 2100
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2101 2102 2103
		prepare_read_connect(con);
		break;

2104
	case CEPH_MSGR_TAG_SEQ:
S
Sage Weil 已提交
2105
	case CEPH_MSGR_TAG_READY:
2106 2107 2108 2109
		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),
2110
			       ceph_pr_addr(&con->peer_addr.in_addr),
2111 2112
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
2113
			reset_connection(con);
2114 2115
			return -1;
		}
2116

2117
		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2118
		con->state = CON_STATE_OPEN;
2119
		con->auth_retry = 0;    /* we authenticated; clear flag */
S
Sage Weil 已提交
2120 2121
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
2122
		con->peer_features = server_feat;
S
Sage Weil 已提交
2123 2124 2125 2126 2127 2128
		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 已提交
2129 2130

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

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

2135 2136 2137 2138 2139 2140
		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 已提交
2141 2142 2143 2144 2145 2146 2147 2148 2149
		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 已提交
2150 2151 2152
		pr_err("process_connect got WAIT as client\n");
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
S
Sage Weil 已提交
2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167

	default:
		pr_err("connect protocol error, will retry\n");
		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)
{
2168 2169
	int size = sizeof (con->in_temp_ack);
	int end = size;
S
Sage Weil 已提交
2170

2171
	return read_partial(con, end, size, &con->in_temp_ack);
S
Sage Weil 已提交
2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190
}

/*
 * 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);
2191
		m->ack_stamp = jiffies;
S
Sage Weil 已提交
2192 2193 2194 2195 2196 2197
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}


2198
static int read_partial_message_section(struct ceph_connection *con,
S
Sage Weil 已提交
2199 2200
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
2201
{
2202
	int ret, left;
2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214

	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;
	}
2215 2216
	if (section->iov_len == sec_len)
		*crc = crc32c(0, section->iov_base, section->iov_len);
S
Sage Weil 已提交
2217

2218 2219
	return 1;
}
S
Sage Weil 已提交
2220

2221 2222 2223
static int read_partial_msg_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->in_msg;
A
Alex Elder 已提交
2224
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
2225
	const bool do_datacrc = !con->msgr->nocrc;
A
Alex Elder 已提交
2226 2227 2228
	struct page *page;
	size_t page_offset;
	size_t length;
A
Alex Elder 已提交
2229
	u32 crc = 0;
2230 2231 2232
	int ret;

	BUG_ON(!msg);
2233
	if (list_empty(&msg->data))
2234
		return -EIO;
2235

A
Alex Elder 已提交
2236 2237
	if (do_datacrc)
		crc = con->in_data_crc;
2238
	while (cursor->resid) {
A
Alex Elder 已提交
2239
		page = ceph_msg_data_next(&msg->cursor, &page_offset, &length,
2240
							NULL);
A
Alex Elder 已提交
2241
		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
A
Alex Elder 已提交
2242 2243 2244 2245
		if (ret <= 0) {
			if (do_datacrc)
				con->in_data_crc = crc;

A
Alex Elder 已提交
2246
			return ret;
A
Alex Elder 已提交
2247
		}
A
Alex Elder 已提交
2248 2249

		if (do_datacrc)
A
Alex Elder 已提交
2250
			crc = ceph_crc32c_page(crc, page, page_offset, ret);
A
Alex Elder 已提交
2251
		(void) ceph_msg_data_advance(&msg->cursor, (size_t)ret);
2252
	}
A
Alex Elder 已提交
2253 2254
	if (do_datacrc)
		con->in_data_crc = crc;
2255 2256 2257 2258

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

S
Sage Weil 已提交
2259 2260 2261
/*
 * read (part of) a message.
 */
A
Alex Elder 已提交
2262 2263
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);

S
Sage Weil 已提交
2264 2265 2266
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
2267 2268
	int size;
	int end;
S
Sage Weil 已提交
2269
	int ret;
2270
	unsigned int front_len, middle_len, data_len;
2271
	bool do_datacrc = !con->msgr->nocrc;
Y
Yan, Zheng 已提交
2272
	bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
2273
	u64 seq;
2274
	u32 crc;
S
Sage Weil 已提交
2275 2276 2277 2278

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

	/* header */
2279 2280 2281
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
2282 2283
	if (ret <= 0)
		return ret;
2284 2285 2286 2287 2288 2289 2290 2291 2292

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

S
Sage Weil 已提交
2293 2294 2295 2296
	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);
2297
	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
S
Sage Weil 已提交
2298 2299 2300 2301 2302
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

2303 2304 2305
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
S
Sage Weil 已提交
2306
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2307
			ENTITY_NAME(con->peer_name),
2308
			ceph_pr_addr(&con->peer_addr.in_addr),
2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320
			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";
		return -EBADMSG;
	}

S
Sage Weil 已提交
2321 2322
	/* allocate message? */
	if (!con->in_msg) {
2323 2324
		int skip = 0;

S
Sage Weil 已提交
2325
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2326
		     front_len, data_len);
2327 2328 2329
		ret = ceph_con_in_msg_alloc(con, &skip);
		if (ret < 0)
			return ret;
2330 2331 2332

		BUG_ON(!con->in_msg ^ skip);
		if (con->in_msg && data_len > con->in_msg->data_length) {
2333
			pr_warn("%s skipping long message (%u > %zd)\n",
2334 2335 2336 2337 2338
				__func__, data_len, con->in_msg->data_length);
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
			skip = 1;
		}
2339
		if (skip) {
S
Sage Weil 已提交
2340
			/* skip this message */
2341
			dout("alloc_msg said skip message\n");
S
Sage Weil 已提交
2342 2343 2344
			con->in_base_pos = -front_len - middle_len - data_len -
				sizeof(m->footer);
			con->in_tag = CEPH_MSGR_TAG_READY;
2345
			con->in_seq++;
S
Sage Weil 已提交
2346 2347
			return 0;
		}
2348

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

2356
		/* prepare for data payload, if any */
2357

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

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

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

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

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

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

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

S
Sage Weil 已提交
2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414
	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;
	}
2415
	if (do_datacrc &&
S
Sage Weil 已提交
2416 2417 2418 2419 2420 2421 2422
	    (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 已提交
2423 2424 2425 2426 2427 2428
	if (need_sign && con->ops->check_message_signature &&
	    con->ops->check_message_signature(con, m)) {
		pr_err("read_partial_message %p signature check failed\n", m);
		return -EBADMSG;
	}

S
Sage Weil 已提交
2429 2430 2431 2432 2433 2434 2435 2436 2437 2438
	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 已提交
2439
	struct ceph_msg *msg;
S
Sage Weil 已提交
2440

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

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

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

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
2456
	     ENTITY_NAME(msg->hdr.src),
S
Sage Weil 已提交
2457 2458 2459 2460 2461 2462
	     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);
2463 2464

	mutex_lock(&con->mutex);
S
Sage Weil 已提交
2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475
}


/*
 * 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;

2476
	dout("try_write start %p state %lu\n", con, con->state);
S
Sage Weil 已提交
2477 2478 2479 2480 2481

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

	/* open the socket first? */
2482 2483 2484
	if (con->state == CON_STATE_PREOPEN) {
		BUG_ON(con->sock);
		con->state = CON_STATE_CONNECTING;
2485

2486
		con_out_kvec_reset(con);
2487
		prepare_write_banner(con);
2488
		prepare_read_banner(con);
S
Sage Weil 已提交
2489

2490
		BUG_ON(con->in_msg);
S
Sage Weil 已提交
2491 2492 2493
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
2494 2495
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
S
Sage Weil 已提交
2496 2497 2498 2499 2500 2501 2502 2503 2504 2505
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
	if (con->out_skip) {
		ret = write_partial_skip(con);
		if (ret <= 0)
2506
			goto out;
S
Sage Weil 已提交
2507 2508 2509 2510
	}
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
		if (ret <= 0)
2511
			goto out;
S
Sage Weil 已提交
2512 2513 2514 2515
	}

	/* msg pages? */
	if (con->out_msg) {
2516 2517 2518 2519 2520 2521
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

2522
		ret = write_partial_message_data(con);
S
Sage Weil 已提交
2523 2524 2525
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2526
			goto out;
S
Sage Weil 已提交
2527
		if (ret < 0) {
2528
			dout("try_write write_partial_message_data err %d\n",
S
Sage Weil 已提交
2529
			     ret);
2530
			goto out;
S
Sage Weil 已提交
2531 2532 2533
		}
	}

2534
do_next:
2535
	if (con->state == CON_STATE_OPEN) {
S
Sage Weil 已提交
2536 2537 2538 2539 2540 2541 2542 2543 2544
		/* 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;
		}
2545
		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
2546 2547 2548 2549 2550 2551
			prepare_write_keepalive(con);
			goto more;
		}
	}

	/* Nothing to do! */
2552
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
2553 2554 2555
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2556
	dout("try_write done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568
	return ret;
}



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

2569 2570 2571 2572 2573
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 已提交
2574 2575
		return 0;

2576
	BUG_ON(!con->sock);
2577

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

2581
	if (con->state == CON_STATE_CONNECTING) {
2582 2583 2584
		dout("try_read connecting\n");
		ret = read_partial_banner(con);
		if (ret <= 0)
2585
			goto out;
2586 2587 2588 2589
		ret = process_banner(con);
		if (ret < 0)
			goto out;

2590
		con->state = CON_STATE_NEGOTIATING;
2591

2592 2593 2594 2595 2596
		/*
		 * Received banner is good, exchange connection info.
		 * Do not reset out_kvec, as sending our banner raced
		 * with receiving peer banner after connect completed.
		 */
2597 2598 2599 2600 2601 2602
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
		prepare_read_connect(con);

		/* Send connection info before awaiting response */
2603 2604 2605
		goto out;
	}

2606
	if (con->state == CON_STATE_NEGOTIATING) {
2607
		dout("try_read negotiating\n");
S
Sage Weil 已提交
2608 2609 2610
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2611 2612 2613
		ret = process_connect(con);
		if (ret < 0)
			goto out;
S
Sage Weil 已提交
2614 2615 2616
		goto more;
	}

2617
	WARN_ON(con->state != CON_STATE_OPEN);
2618

S
Sage Weil 已提交
2619 2620 2621 2622 2623 2624
	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
A
Alex Elder 已提交
2625 2626 2627
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

S
Sage Weil 已提交
2628 2629 2630
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2631
			goto out;
S
Sage Weil 已提交
2632 2633 2634 2635 2636 2637 2638 2639 2640 2641
		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)
2642
			goto out;
S
Sage Weil 已提交
2643 2644 2645 2646 2647 2648 2649 2650 2651
		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;
		case CEPH_MSGR_TAG_CLOSE:
2652 2653
			con_close_socket(con);
			con->state = CON_STATE_CLOSED;
2654
			goto out;
S
Sage Weil 已提交
2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665
		default:
			goto bad_tag;
		}
	}
	if (con->in_tag == CEPH_MSGR_TAG_MSG) {
		ret = read_partial_message(con);
		if (ret <= 0) {
			switch (ret) {
			case -EBADMSG:
				con->error_msg = "bad crc";
				ret = -EIO;
2666
				break;
S
Sage Weil 已提交
2667 2668
			case -EIO:
				con->error_msg = "io error";
2669
				break;
S
Sage Weil 已提交
2670
			}
2671
			goto out;
S
Sage Weil 已提交
2672 2673 2674 2675
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
2676 2677
		if (con->state == CON_STATE_OPEN)
			prepare_read_tag(con);
S
Sage Weil 已提交
2678 2679
		goto more;
	}
2680 2681 2682 2683 2684 2685
	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 已提交
2686 2687
		ret = read_partial_ack(con);
		if (ret <= 0)
2688
			goto out;
S
Sage Weil 已提交
2689 2690 2691 2692 2693
		process_ack(con);
		goto more;
	}

out:
2694
	dout("try_read done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705
	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 已提交
2706 2707 2708
 * 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 已提交
2709
 */
A
Alex Elder 已提交
2710
static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
S
Sage Weil 已提交
2711 2712
{
	if (!con->ops->get(con)) {
A
Alex Elder 已提交
2713 2714
		dout("%s %p ref count 0\n", __func__, con);
		return -ENOENT;
S
Sage Weil 已提交
2715 2716
	}

A
Alex Elder 已提交
2717 2718
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
		dout("%s %p - already queued\n", __func__, con);
S
Sage Weil 已提交
2719
		con->ops->put(con);
A
Alex Elder 已提交
2720
		return -EBUSY;
S
Sage Weil 已提交
2721
	}
A
Alex Elder 已提交
2722 2723 2724 2725 2726 2727 2728 2729

	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 已提交
2730 2731
}

2732 2733 2734 2735 2736 2737 2738 2739
static void cancel_con(struct ceph_connection *con)
{
	if (cancel_delayed_work(&con->work)) {
		dout("%s %p\n", __func__, con);
		con->ops->put(con);
	}
}

2740 2741
static bool con_sock_closed(struct ceph_connection *con)
{
2742
	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753 2754 2755 2756 2757
		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:
2758
		pr_warn("%s con %p unrecognized state %lu\n",
2759 2760 2761 2762 2763 2764 2765 2766 2767 2768
			__func__, con, con->state);
		con->error_msg = "unrecognized con state";
		BUG();
		break;
	}
#undef CASE

	return true;
}

2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786
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;
}

2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803
/* 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 已提交
2804 2805 2806 2807 2808 2809 2810
/*
 * Do some work on a connection.  Drop a connection ref when we're done.
 */
static void con_work(struct work_struct *work)
{
	struct ceph_connection *con = container_of(work, struct ceph_connection,
						   work.work);
2811
	unsigned long pflags = current->flags;
2812
	bool fault;
S
Sage Weil 已提交
2813

2814 2815
	current->flags |= PF_MEMALLOC;

S
Sage Weil 已提交
2816
	mutex_lock(&con->mutex);
2817 2818
	while (true) {
		int ret;
S
Sage Weil 已提交
2819

2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840
		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);
		}
2841

2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859
		ret = try_read(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
			con->error_msg = "socket error on read";
			fault = true;
			break;
		}

		ret = try_write(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
			con->error_msg = "socket error on write";
			fault = true;
		}

		break;	/* If we make it to here, we're done */
2860
	}
2861 2862
	if (fault)
		con_fault(con);
S
Sage Weil 已提交
2863
	mutex_unlock(&con->mutex);
2864

2865 2866 2867 2868
	if (fault)
		con_fault_finish(con);

	con->ops->put(con);
2869 2870

	tsk_restore_flags(current, pflags, PF_MEMALLOC);
S
Sage Weil 已提交
2871 2872 2873 2874 2875 2876
}

/*
 * Generic error/fault handler.  A retry mechanism is used with
 * exponential backoff
 */
2877
static void con_fault(struct ceph_connection *con)
S
Sage Weil 已提交
2878
{
2879 2880
	pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
		ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
S
Sage Weil 已提交
2881
	dout("fault %p state %lu to peer %s\n",
2882
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2883

2884
	WARN_ON(con->state != CON_STATE_CONNECTING &&
2885 2886
	       con->state != CON_STATE_NEGOTIATING &&
	       con->state != CON_STATE_OPEN);
2887

S
Sage Weil 已提交
2888
	con_close_socket(con);
S
Sage Weil 已提交
2889

2890
	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2891 2892
		dout("fault on LOSSYTX channel, marking CLOSED\n");
		con->state = CON_STATE_CLOSED;
2893
		return;
2894 2895
	}

S
Sage Weil 已提交
2896
	if (con->in_msg) {
2897 2898
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
S
Sage Weil 已提交
2899 2900
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
S
Sage Weil 已提交
2901
		con->ops->put(con);
S
Sage Weil 已提交
2902
	}
S
Sage Weil 已提交
2903

2904 2905
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2906

S
Sage Weil 已提交
2907 2908 2909
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2910
	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
2911
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2912
		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2913
		con->state = CON_STATE_STANDBY;
2914 2915
	} else {
		/* retry after a delay. */
2916
		con->state = CON_STATE_PREOPEN;
2917 2918 2919 2920
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
2921
		con_flag_set(con, CON_FLAG_BACKOFF);
A
Alex Elder 已提交
2922
		queue_con(con);
S
Sage Weil 已提交
2923 2924 2925 2926 2927 2928
	}
}



/*
2929
 * initialize a new messenger instance
S
Sage Weil 已提交
2930
 */
2931 2932
void ceph_messenger_init(struct ceph_messenger *msgr,
			struct ceph_entity_addr *myaddr,
2933 2934
			u64 supported_features,
			u64 required_features,
C
Chaitanya Huilgol 已提交
2935 2936
			bool nocrc,
			bool tcp_nodelay)
S
Sage Weil 已提交
2937
{
2938 2939 2940
	msgr->supported_features = supported_features;
	msgr->required_features = required_features;

S
Sage Weil 已提交
2941 2942 2943 2944 2945 2946
	spin_lock_init(&msgr->global_seq_lock);

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

	/* select a random nonce */
2947
	msgr->inst.addr.type = 0;
S
Sage Weil 已提交
2948
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2949
	encode_my_addr(msgr);
2950
	msgr->nocrc = nocrc;
C
Chaitanya Huilgol 已提交
2951
	msgr->tcp_nodelay = tcp_nodelay;
S
Sage Weil 已提交
2952

2953
	atomic_set(&msgr->stopping, 0);
S
Sage Weil 已提交
2954

2955
	dout("%s %p\n", __func__, msgr);
S
Sage Weil 已提交
2956
}
2957
EXPORT_SYMBOL(ceph_messenger_init);
S
Sage Weil 已提交
2958

S
Sage Weil 已提交
2959 2960 2961
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
2962
	if (con->state == CON_STATE_STANDBY) {
S
Sage Weil 已提交
2963
		dout("clear_standby %p and ++connect_seq\n", con);
2964
		con->state = CON_STATE_PREOPEN;
S
Sage Weil 已提交
2965
		con->connect_seq++;
2966 2967
		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
S
Sage Weil 已提交
2968 2969 2970
	}
}

S
Sage Weil 已提交
2971 2972 2973 2974 2975 2976
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	/* set src+dst */
2977
	msg->hdr.src = con->msgr->inst.name;
2978
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2979 2980
	msg->needs_out_seq = true;

2981
	mutex_lock(&con->mutex);
2982

2983
	if (con->state == CON_STATE_CLOSED) {
2984 2985 2986 2987 2988 2989
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		mutex_unlock(&con->mutex);
		return;
	}

2990
	BUG_ON(msg->con != NULL);
S
Sage Weil 已提交
2991
	msg->con = con->ops->get(con);
2992 2993
	BUG_ON(msg->con == NULL);

S
Sage Weil 已提交
2994 2995 2996 2997 2998 2999 3000 3001
	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));
3002 3003

	clear_standby(con);
3004
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3005 3006 3007

	/* if there wasn't anything waiting to send before, queue
	 * new work */
3008
	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
3009 3010
		queue_con(con);
}
3011
EXPORT_SYMBOL(ceph_con_send);
S
Sage Weil 已提交
3012 3013 3014 3015

/*
 * Revoke a message that was previously queued for send
 */
3016
void ceph_msg_revoke(struct ceph_msg *msg)
S
Sage Weil 已提交
3017
{
3018 3019 3020 3021 3022
	struct ceph_connection *con = msg->con;

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

3023
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3024
	if (!list_empty(&msg->list_head)) {
3025
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
S
Sage Weil 已提交
3026
		list_del_init(&msg->list_head);
3027
		BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
3028
		msg->con->ops->put(msg->con);
3029
		msg->con = NULL;
S
Sage Weil 已提交
3030
		msg->hdr.seq = 0;
3031

S
Sage Weil 已提交
3032
		ceph_msg_put(msg);
S
Sage Weil 已提交
3033 3034
	}
	if (con->out_msg == msg) {
3035
		dout("%s %p msg %p - was sending\n", __func__, con, msg);
S
Sage Weil 已提交
3036
		con->out_msg = NULL;
S
Sage Weil 已提交
3037 3038 3039 3040
		if (con->out_kvec_is_msg) {
			con->out_skip = con->out_kvec_bytes;
			con->out_kvec_is_msg = false;
		}
S
Sage Weil 已提交
3041
		msg->hdr.seq = 0;
3042 3043

		ceph_msg_put(msg);
S
Sage Weil 已提交
3044
	}
3045
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3046 3047
}

3048
/*
3049
 * Revoke a message that we may be reading data into
3050
 */
3051
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
3052
{
3053 3054 3055 3056 3057 3058 3059 3060 3061 3062
	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;
3063
	mutex_lock(&con->mutex);
3064
	if (con->in_msg == msg) {
3065 3066 3067
		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);
3068 3069

		/* skip rest of message */
3070 3071
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
3072
				sizeof(struct ceph_msg_header) -
3073 3074 3075
				front_len -
				middle_len -
				data_len -
3076 3077 3078 3079
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
3080
		con->in_seq++;
3081
	} else {
3082 3083
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
3084 3085 3086 3087
	}
	mutex_unlock(&con->mutex);
}

S
Sage Weil 已提交
3088 3089 3090 3091 3092
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
S
Sage Weil 已提交
3093
	dout("con_keepalive %p\n", con);
3094
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3095
	clear_standby(con);
3096
	mutex_unlock(&con->mutex);
3097 3098
	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 已提交
3099 3100
		queue_con(con);
}
3101
EXPORT_SYMBOL(ceph_con_keepalive);
S
Sage Weil 已提交
3102

3103
static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type)
A
Alex Elder 已提交
3104
{
3105 3106 3107 3108 3109
	struct ceph_msg_data *data;

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

3110
	data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS);
3111 3112
	if (data)
		data->type = type;
3113
	INIT_LIST_HEAD(&data->links);
3114 3115 3116 3117 3118 3119 3120 3121 3122

	return data;
}

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

3123
	WARN_ON(!list_empty(&data->links));
Y
Yan, Zheng 已提交
3124
	if (data->type == CEPH_MSG_DATA_PAGELIST)
3125
		ceph_pagelist_release(data->pagelist);
3126
	kmem_cache_free(ceph_msg_data_cache, data);
A
Alex Elder 已提交
3127 3128
}

3129
void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
3130
		size_t length, size_t alignment)
3131
{
3132 3133
	struct ceph_msg_data *data;

3134 3135
	BUG_ON(!pages);
	BUG_ON(!length);
3136 3137 3138 3139 3140 3141

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

3143 3144
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3145
}
3146
EXPORT_SYMBOL(ceph_msg_data_add_pages);
S
Sage Weil 已提交
3147

3148
void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
3149 3150
				struct ceph_pagelist *pagelist)
{
3151 3152
	struct ceph_msg_data *data;

3153 3154
	BUG_ON(!pagelist);
	BUG_ON(!pagelist->length);
3155

3156 3157 3158 3159
	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
	BUG_ON(!data);
	data->pagelist = pagelist;

3160 3161
	list_add_tail(&data->links, &msg->data);
	msg->data_length += pagelist->length;
3162
}
3163
EXPORT_SYMBOL(ceph_msg_data_add_pagelist);
3164

3165
#ifdef	CONFIG_BLOCK
3166
void ceph_msg_data_add_bio(struct ceph_msg *msg, struct bio *bio,
A
Alex Elder 已提交
3167
		size_t length)
3168
{
3169 3170
	struct ceph_msg_data *data;

3171
	BUG_ON(!bio);
3172

3173 3174 3175
	data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
	BUG_ON(!data);
	data->bio = bio;
A
Alex Elder 已提交
3176
	data->bio_length = length;
3177

3178 3179
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3180
}
3181
EXPORT_SYMBOL(ceph_msg_data_add_bio);
3182
#endif	/* CONFIG_BLOCK */
3183

S
Sage Weil 已提交
3184 3185 3186 3187
/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
3188 3189
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
S
Sage Weil 已提交
3190 3191 3192
{
	struct ceph_msg *m;

3193
	m = kmem_cache_zalloc(ceph_msg_cache, flags);
S
Sage Weil 已提交
3194 3195 3196 3197
	if (m == NULL)
		goto out;

	m->hdr.type = cpu_to_le16(type);
3198
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
S
Sage Weil 已提交
3199
	m->hdr.front_len = cpu_to_le32(front_len);
3200

3201 3202
	INIT_LIST_HEAD(&m->list_head);
	kref_init(&m->kref);
3203
	INIT_LIST_HEAD(&m->data);
3204

S
Sage Weil 已提交
3205 3206
	/* front */
	if (front_len) {
3207
		m->front.iov_base = ceph_kvmalloc(front_len, flags);
S
Sage Weil 已提交
3208
		if (m->front.iov_base == NULL) {
3209
			dout("ceph_msg_new can't allocate %d bytes\n",
S
Sage Weil 已提交
3210 3211 3212 3213 3214 3215
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
3216
	m->front_alloc_len = m->front.iov_len = front_len;
S
Sage Weil 已提交
3217

S
Sage Weil 已提交
3218
	dout("ceph_msg_new %p front %d\n", m, front_len);
S
Sage Weil 已提交
3219 3220 3221 3222 3223
	return m;

out2:
	ceph_msg_put(m);
out:
3224 3225 3226
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
3227
		WARN_ON(1);
3228 3229 3230 3231
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
3232
	return NULL;
S
Sage Weil 已提交
3233
}
3234
EXPORT_SYMBOL(ceph_msg_new);
S
Sage Weil 已提交
3235 3236 3237 3238 3239 3240 3241 3242

/*
 * 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.
 */
3243
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
S
Sage Weil 已提交
3244 3245 3246 3247 3248 3249 3250 3251 3252
{
	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 已提交
3253
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
S
Sage Weil 已提交
3254 3255 3256 3257 3258
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

3259
/*
A
Alex Elder 已提交
3260 3261 3262 3263
 * 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.
 *
3264 3265 3266 3267 3268 3269 3270 3271 3272
 * 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
3273
 */
3274
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3275
{
3276
	struct ceph_msg_header *hdr = &con->in_hdr;
3277
	int middle_len = le32_to_cpu(hdr->middle_len);
3278
	struct ceph_msg *msg;
3279
	int ret = 0;
3280

A
Alex Elder 已提交
3281
	BUG_ON(con->in_msg != NULL);
3282
	BUG_ON(!con->ops->alloc_msg);
3283

3284 3285 3286 3287 3288
	mutex_unlock(&con->mutex);
	msg = con->ops->alloc_msg(con, hdr, skip);
	mutex_lock(&con->mutex);
	if (con->state != CON_STATE_OPEN) {
		if (msg)
3289
			ceph_msg_put(msg);
3290 3291
		return -EAGAIN;
	}
3292 3293 3294
	if (msg) {
		BUG_ON(*skip);
		con->in_msg = msg;
S
Sage Weil 已提交
3295
		con->in_msg->con = con->ops->get(con);
3296
		BUG_ON(con->in_msg->con == NULL);
3297 3298 3299 3300 3301 3302 3303 3304 3305 3306
	} 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;
		con->error_msg = "error allocating memory for incoming message";

3307
		return -ENOMEM;
3308
	}
A
Alex Elder 已提交
3309
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3310

A
Alex Elder 已提交
3311 3312
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
3313
		if (ret < 0) {
A
Alex Elder 已提交
3314 3315
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
3316 3317
		}
	}
3318

3319
	return ret;
3320 3321
}

S
Sage Weil 已提交
3322 3323 3324 3325

/*
 * Free a generically kmalloc'd message.
 */
3326
static void ceph_msg_free(struct ceph_msg *m)
S
Sage Weil 已提交
3327
{
3328
	dout("%s %p\n", __func__, m);
I
Ilya Dryomov 已提交
3329
	kvfree(m->front.iov_base);
3330
	kmem_cache_free(ceph_msg_cache, m);
S
Sage Weil 已提交
3331 3332
}

3333
static void ceph_msg_release(struct kref *kref)
S
Sage Weil 已提交
3334 3335
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
3336 3337 3338
	LIST_HEAD(data);
	struct list_head *links;
	struct list_head *next;
S
Sage Weil 已提交
3339

3340
	dout("%s %p\n", __func__, m);
S
Sage Weil 已提交
3341 3342 3343 3344 3345 3346
	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 已提交
3347
	}
3348 3349 3350 3351 3352 3353 3354 3355 3356

	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 已提交
3357
	m->data_length = 0;
3358

S
Sage Weil 已提交
3359 3360 3361
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
3362 3363 3364 3365 3366 3367 3368 3369 3370 3371 3372 3373 3374 3375 3376 3377 3378
		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 已提交
3379
}
3380
EXPORT_SYMBOL(ceph_msg_put);
3381 3382 3383

void ceph_msg_dump(struct ceph_msg *msg)
{
3384 3385
	pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg,
		 msg->front_alloc_len, msg->data_length);
3386 3387 3388 3389 3390 3391 3392 3393 3394 3395 3396 3397 3398 3399 3400
	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);
}
3401
EXPORT_SYMBOL(ceph_msg_dump);