messenger.c 81.3 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>

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

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

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

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

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

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

S
Sage Weil 已提交
155 156 157 158 159
/* 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;

160 161 162 163
#ifdef CONFIG_LOCKDEP
static struct lock_class_key socket_class;
#endif

A
Alex Elder 已提交
164 165 166 167 168
/*
 * 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 已提交
169 170 171

static void queue_con(struct ceph_connection *con);
static void con_work(struct work_struct *);
172
static void con_fault(struct ceph_connection *con);
S
Sage Weil 已提交
173 174

/*
175 176
 * Nicely render a sockaddr as a string.  An array of formatted
 * strings is used, to approximate reentrancy.
S
Sage Weil 已提交
177
 */
178 179 180 181 182 183 184
#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 已提交
185

186 187
static struct page *zero_page;		/* used in certain error cases */

188
const char *ceph_pr_addr(const struct sockaddr_storage *ss)
S
Sage Weil 已提交
189 190 191
{
	int i;
	char *s;
A
Alex Elder 已提交
192 193
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
S
Sage Weil 已提交
194

195
	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
S
Sage Weil 已提交
196 197 198 199
	s = addr_str[i];

	switch (ss->ss_family) {
	case AF_INET:
A
Alex Elder 已提交
200 201
		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
			 ntohs(in4->sin_port));
S
Sage Weil 已提交
202 203 204
		break;

	case AF_INET6:
A
Alex Elder 已提交
205 206
		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
			 ntohs(in6->sin6_port));
S
Sage Weil 已提交
207 208 209
		break;

	default:
A
Alex Elder 已提交
210 211
		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
			 ss->ss_family);
S
Sage Weil 已提交
212 213 214 215
	}

	return s;
}
216
EXPORT_SYMBOL(ceph_pr_addr);
S
Sage Weil 已提交
217

218 219 220 221 222 223
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 已提交
224 225 226
/*
 * work queue for all reading and writing to/from the socket.
 */
A
Alex Elder 已提交
227
static struct workqueue_struct *ceph_msgr_wq;
S
Sage Weil 已提交
228

A
Alex Elder 已提交
229
static void _ceph_msgr_exit(void)
230
{
A
Alex Elder 已提交
231
	if (ceph_msgr_wq) {
232
		destroy_workqueue(ceph_msgr_wq);
A
Alex Elder 已提交
233 234
		ceph_msgr_wq = NULL;
	}
235 236 237 238 239 240 241

	BUG_ON(zero_page == NULL);
	kunmap(zero_page);
	page_cache_release(zero_page);
	zero_page = NULL;
}

242
int ceph_msgr_init(void)
S
Sage Weil 已提交
243
{
244 245 246 247
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
	page_cache_get(zero_page);

248
	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_NON_REENTRANT, 0);
249 250
	if (ceph_msgr_wq)
		return 0;
251

252 253
	pr_err("msgr_init failed to create workqueue\n");
	_ceph_msgr_exit();
254

255
	return -ENOMEM;
S
Sage Weil 已提交
256
}
257
EXPORT_SYMBOL(ceph_msgr_init);
S
Sage Weil 已提交
258 259 260

void ceph_msgr_exit(void)
{
261 262
	BUG_ON(ceph_msgr_wq == NULL);

263
	_ceph_msgr_exit();
S
Sage Weil 已提交
264
}
265
EXPORT_SYMBOL(ceph_msgr_exit);
S
Sage Weil 已提交
266

Y
Yehuda Sadeh 已提交
267
void ceph_msgr_flush(void)
268 269 270
{
	flush_workqueue(ceph_msgr_wq);
}
271
EXPORT_SYMBOL(ceph_msgr_flush);
272

273 274 275 276 277 278 279 280 281
/* 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);
282 283
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
284 285 286 287 288 289 290 291 292
}

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);
293 294
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTING);
295 296 297 298 299 300 301 302 303
}

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);
304 305
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTED);
306 307 308 309 310 311 312 313 314 315 316
}

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);
317 318
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSING);
319 320 321 322 323 324 325 326
}

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 &&
327
		    old_state != CON_SOCK_STATE_CLOSING &&
328 329
		    old_state != CON_SOCK_STATE_CONNECTING &&
		    old_state != CON_SOCK_STATE_CLOSED))
330
		printk("%s: unexpected old state %d\n", __func__, old_state);
331 332
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
333
}
334

S
Sage Weil 已提交
335 336 337 338 339
/*
 * socket callback functions
 */

/* data available on socket, or listen socket received a connect */
A
Alex Elder 已提交
340
static void ceph_sock_data_ready(struct sock *sk, int count_unused)
S
Sage Weil 已提交
341
{
A
Alex Elder 已提交
342
	struct ceph_connection *con = sk->sk_user_data;
343 344 345
	if (atomic_read(&con->msgr->stopping)) {
		return;
	}
A
Alex Elder 已提交
346

S
Sage Weil 已提交
347
	if (sk->sk_state != TCP_CLOSE_WAIT) {
A
Alex Elder 已提交
348
		dout("%s on %p state = %lu, queueing work\n", __func__,
S
Sage Weil 已提交
349 350 351 352 353 354
		     con, con->state);
		queue_con(con);
	}
}

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

359 360
	/* 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 已提交
361
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
362 363 364 365
	 * 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().
	 */
366
	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
367
		if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
A
Alex Elder 已提交
368
			dout("%s %p queueing write work\n", __func__, con);
369 370 371
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
S
Sage Weil 已提交
372
	} else {
A
Alex Elder 已提交
373
		dout("%s %p nothing to write\n", __func__, con);
S
Sage Weil 已提交
374 375 376 377
	}
}

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

A
Alex Elder 已提交
382
	dout("%s %p state = %lu sk_state = %u\n", __func__,
S
Sage Weil 已提交
383 384 385 386
	     con, con->state, sk->sk_state);

	switch (sk->sk_state) {
	case TCP_CLOSE:
A
Alex Elder 已提交
387
		dout("%s TCP_CLOSE\n", __func__);
S
Sage Weil 已提交
388
	case TCP_CLOSE_WAIT:
A
Alex Elder 已提交
389
		dout("%s TCP_CLOSE_WAIT\n", __func__);
390
		con_sock_state_closing(con);
391
		con_flag_set(con, CON_FLAG_SOCK_CLOSED);
392
		queue_con(con);
S
Sage Weil 已提交
393 394
		break;
	case TCP_ESTABLISHED:
A
Alex Elder 已提交
395
		dout("%s TCP_ESTABLISHED\n", __func__);
396
		con_sock_state_connected(con);
S
Sage Weil 已提交
397 398
		queue_con(con);
		break;
A
Alex Elder 已提交
399 400
	default:	/* Everything else is uninteresting */
		break;
S
Sage Weil 已提交
401 402 403 404 405 406 407 408 409 410
	}
}

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
A
Alex Elder 已提交
411
	sk->sk_user_data = con;
A
Alex Elder 已提交
412 413 414
	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 已提交
415 416 417 418 419 420 421 422 423 424
}


/*
 * socket helpers
 */

/*
 * initiate connection to a remote socket.
 */
425
static int ceph_tcp_connect(struct ceph_connection *con)
S
Sage Weil 已提交
426
{
S
Sage Weil 已提交
427
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
S
Sage Weil 已提交
428 429 430 431
	struct socket *sock;
	int ret;

	BUG_ON(con->sock);
S
Sage Weil 已提交
432 433
	ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
			       IPPROTO_TCP, &sock);
S
Sage Weil 已提交
434
	if (ret)
435
		return ret;
S
Sage Weil 已提交
436 437
	sock->sk->sk_allocation = GFP_NOFS;

438 439 440 441
#ifdef CONFIG_LOCKDEP
	lockdep_set_class(&sock->sk->sk_lock, &socket_class);
#endif

S
Sage Weil 已提交
442 443
	set_sock_callbacks(sock, con);

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

446
	con_sock_state_connecting(con);
S
Sage Weil 已提交
447 448
	ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
				 O_NONBLOCK);
S
Sage Weil 已提交
449 450
	if (ret == -EINPROGRESS) {
		dout("connect %s EINPROGRESS sk_state = %u\n",
451
		     ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
452
		     sock->sk->sk_state);
453
	} else if (ret < 0) {
S
Sage Weil 已提交
454
		pr_err("connect %s error %d\n",
455
		       ceph_pr_addr(&con->peer_addr.in_addr), ret);
S
Sage Weil 已提交
456 457 458
		sock_release(sock);
		con->error_msg = "connect error";

459
		return ret;
460 461
	}
	con->sock = sock;
462
	return 0;
S
Sage Weil 已提交
463 464 465 466 467 468
}

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

471 472 473 474
	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
475 476
}

477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492
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 已提交
493 494 495 496 497 498 499 500
/*
 * 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 };
501
	int r;
S
Sage Weil 已提交
502 503 504 505 506 507

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

508 509 510 511
	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
512 513
}

514
static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
515
		     int offset, size_t size, bool more)
516 517 518 519 520 521 522 523 524 525 526
{
	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;
}

S
Sage Weil 已提交
527 528 529 530 531 532

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

	dout("con_close_socket on %p sock %p\n", con, con->sock);
536 537 538 539 540
	if (con->sock) {
		rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
		sock_release(con->sock);
		con->sock = NULL;
	}
541 542

	/*
S
Sage Weil 已提交
543
	 * Forcibly clear the SOCK_CLOSED flag.  It gets set
544 545 546 547
	 * independent of the connection mutex, and we could have
	 * received a socket close event before we had the chance to
	 * shut the socket down.
	 */
548
	con_flag_clear(con, CON_FLAG_SOCK_CLOSED);
549

550
	con_sock_state_closed(con);
S
Sage Weil 已提交
551 552 553 554 555 556 557 558 559 560
	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);
561
	BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
562
	msg->con->ops->put(msg->con);
563 564
	msg->con = NULL;

S
Sage Weil 已提交
565 566 567 568 569 570 571 572 573 574 575 576 577 578 579
	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 */
580
	dout("reset_connection %p\n", con);
S
Sage Weil 已提交
581 582 583
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

584
	if (con->in_msg) {
585 586
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
587 588
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
S
Sage Weil 已提交
589
		con->ops->put(con);
590 591
	}

S
Sage Weil 已提交
592 593
	con->connect_seq = 0;
	con->out_seq = 0;
594 595 596 597
	if (con->out_msg) {
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
S
Sage Weil 已提交
598
	con->in_seq = 0;
599
	con->in_seq_acked = 0;
S
Sage Weil 已提交
600 601 602 603 604 605 606
}

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

612 613 614 615
	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);
616

S
Sage Weil 已提交
617
	reset_connection(con);
S
Sage Weil 已提交
618
	con->peer_global_seq = 0;
619
	cancel_delayed_work(&con->work);
620
	con_close_socket(con);
621
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
622
}
623
EXPORT_SYMBOL(ceph_con_close);
S
Sage Weil 已提交
624 625 626 627

/*
 * Reopen a closed connection, with a new peer address.
 */
628 629 630
void ceph_con_open(struct ceph_connection *con,
		   __u8 entity_type, __u64 entity_num,
		   struct ceph_entity_addr *addr)
S
Sage Weil 已提交
631
{
632
	mutex_lock(&con->mutex);
633
	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
634

635
	WARN_ON(con->state != CON_STATE_CLOSED);
636
	con->state = CON_STATE_PREOPEN;
637

638 639 640
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

S
Sage Weil 已提交
641
	memcpy(&con->peer_addr, addr, sizeof(*addr));
642
	con->delay = 0;      /* reset backoff memory */
643
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
644 645
	queue_con(con);
}
646
EXPORT_SYMBOL(ceph_con_open);
S
Sage Weil 已提交
647

648 649 650 651 652 653 654 655
/*
 * return true if this connection ever successfully opened
 */
bool ceph_con_opened(struct ceph_connection *con)
{
	return con->connect_seq > 0;
}

S
Sage Weil 已提交
656 657 658
/*
 * initialize a new connection.
 */
659 660
void ceph_con_init(struct ceph_connection *con, void *private,
	const struct ceph_connection_operations *ops,
661
	struct ceph_messenger *msgr)
S
Sage Weil 已提交
662 663 664
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
665 666
	con->private = private;
	con->ops = ops;
S
Sage Weil 已提交
667
	con->msgr = msgr;
668 669 670

	con_sock_state_init(con);

671
	mutex_init(&con->mutex);
S
Sage Weil 已提交
672 673 674
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
	INIT_DELAYED_WORK(&con->work, con_work);
675

676
	con->state = CON_STATE_CLOSED;
S
Sage Weil 已提交
677
}
678
EXPORT_SYMBOL(ceph_con_init);
S
Sage Weil 已提交
679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696


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

697
static void con_out_kvec_reset(struct ceph_connection *con)
698 699 700 701 702 703
{
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

704
static void con_out_kvec_add(struct ceph_connection *con,
705 706 707 708 709 710 711 712 713 714 715 716
				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 已提交
717

718
#ifdef CONFIG_BLOCK
719 720 721 722 723 724

/*
 * 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.
 */
725 726
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data *data,
					size_t length)
727 728 729 730 731 732 733 734 735 736
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct bio *bio;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

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

737
	cursor->resid = length;
738 739 740
	cursor->bio = bio;
	cursor->vector_index = 0;
	cursor->vector_offset = 0;
741
	cursor->last_piece = length <= bio->bi_io_vec[0].bv_len;
742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764
}

static struct page *ceph_msg_data_bio_next(struct ceph_msg_data *data,
						size_t *page_offset,
						size_t *length)
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct bio *bio;
	struct bio_vec *bio_vec;
	unsigned int index;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

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

	index = cursor->vector_index;
	BUG_ON(index >= (unsigned int) bio->bi_vcnt);

	bio_vec = &bio->bi_io_vec[index];
	BUG_ON(cursor->vector_offset >= bio_vec->bv_len);
	*page_offset = (size_t) (bio_vec->bv_offset + cursor->vector_offset);
	BUG_ON(*page_offset >= PAGE_SIZE);
765 766 767 768
	if (cursor->last_piece) /* pagelist offset is always 0 */
		*length = cursor->resid;
	else
		*length = (size_t) (bio_vec->bv_len - cursor->vector_offset);
769
	BUG_ON(*length > PAGE_SIZE);
770
	BUG_ON(*length > cursor->resid);
771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792

	return bio_vec->bv_page;
}

static bool ceph_msg_data_bio_advance(struct ceph_msg_data *data, size_t bytes)
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct bio *bio;
	struct bio_vec *bio_vec;
	unsigned int index;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

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

	index = cursor->vector_index;
	BUG_ON(index >= (unsigned int) bio->bi_vcnt);
	bio_vec = &bio->bi_io_vec[index];

	/* Advance the cursor offset */

793 794
	BUG_ON(cursor->resid < bytes);
	cursor->resid -= bytes;
795 796 797
	cursor->vector_offset += bytes;
	if (cursor->vector_offset < bio_vec->bv_len)
		return false;	/* more bytes to process in this segment */
798
	BUG_ON(cursor->vector_offset != bio_vec->bv_len);
799 800 801

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

802
	if (++index == (unsigned int) bio->bi_vcnt) {
803
		bio = bio->bi_next;
804
		index = 0;
805
	}
806 807
	cursor->bio = bio;
	cursor->vector_index = index;
808 809
	cursor->vector_offset = 0;

810 811 812 813 814
	if (!cursor->last_piece) {
		BUG_ON(!cursor->resid);
		BUG_ON(!bio);
		/* A short read is OK, so use <= rather than == */
		if (cursor->resid <= bio->bi_io_vec[index].bv_len)
815
			cursor->last_piece = true;
816
	}
817 818 819

	return true;
}
820 821
#endif

822 823 824 825
/*
 * For a page array, a piece comes from the first page in the array
 * that has not already been fully consumed.
 */
826 827
static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data *data,
					size_t length)
828 829 830 831 832 833 834 835
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	int page_count;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

	BUG_ON(!data->pages);
	BUG_ON(!data->length);
836
	BUG_ON(length != data->length);
837

838
	cursor->resid = length;
839 840 841
	page_count = calc_pages_for(data->alignment, (u64)data->length);
	cursor->page_offset = data->alignment & ~PAGE_MASK;
	cursor->page_index = 0;
842
	BUG_ON(page_count > (int) USHRT_MAX);
843
	cursor->page_count = (unsigned short) page_count;
844
	cursor->last_piece = length <= PAGE_SIZE;
845 846 847 848 849 850 851 852 853 854 855 856 857 858
}

static struct page *ceph_msg_data_pages_next(struct ceph_msg_data *data,
					size_t *page_offset,
					size_t *length)
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;

	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;
859
	if (cursor->last_piece)
860
		*length = cursor->resid;
861
	else
862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887
		*length = PAGE_SIZE - *page_offset;

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

static bool ceph_msg_data_pages_advance(struct ceph_msg_data *data,
						size_t bytes)
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

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

	/* Advance the cursor page offset */

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

	/* Move on to the next page */

	BUG_ON(cursor->page_index >= cursor->page_count);
	cursor->page_offset = 0;
	cursor->page_index++;
888
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
889 890 891 892

	return true;
}

893
/*
894 895
 * 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.
896
 */
897 898
static void ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data *data,
					size_t length)
899 900 901 902 903
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct ceph_pagelist *pagelist;
	struct page *page;

904
	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
905 906 907

	pagelist = data->pagelist;
	BUG_ON(!pagelist);
908 909 910
	BUG_ON(length != pagelist->length);

	if (!length)
911 912 913 914 915
		return;		/* pagelist can be assigned but empty */

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

916
	cursor->resid = length;
917 918
	cursor->page = page;
	cursor->offset = 0;
919
	cursor->last_piece = length <= PAGE_SIZE;
920 921
}

922
static struct page *ceph_msg_data_pagelist_next(struct ceph_msg_data *data,
923
						size_t *page_offset,
924
						size_t *length)
925 926 927 928 929 930 931 932 933 934
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(!cursor->page);
935
	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
936 937

	*page_offset = cursor->offset & ~PAGE_MASK;
938 939 940 941
	if (cursor->last_piece) /* pagelist offset is always 0 */
		*length = cursor->resid;
	else
		*length = PAGE_SIZE - *page_offset;
942 943 944 945

	return data->cursor.page;
}

946 947
static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data *data,
						size_t bytes)
948 949 950 951 952 953 954 955
{
	struct ceph_msg_data_cursor *cursor = &data->cursor;
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
958 959 960 961
	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);

	/* Advance the cursor offset */

962
	cursor->resid -= bytes;
963 964 965 966 967 968 969 970 971
	cursor->offset += bytes;
	/* pagelist offset is always 0 */
	if (!bytes || cursor->offset & ~PAGE_MASK)
		return false;	/* more bytes to process in the current page */

	/* Move on to the next page */

	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
	cursor->page = list_entry_next(cursor->page, lru);
972
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
973 974 975 976

	return true;
}

977 978 979 980 981 982 983 984
/*
 * 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.
 */
985 986
static void ceph_msg_data_cursor_init(struct ceph_msg_data *data,
					size_t length)
987 988 989
{
	switch (data->type) {
	case CEPH_MSG_DATA_PAGELIST:
990
		ceph_msg_data_pagelist_cursor_init(data, length);
991
		break;
992
	case CEPH_MSG_DATA_PAGES:
993
		ceph_msg_data_pages_cursor_init(data, length);
994
		break;
995 996
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
997
		ceph_msg_data_bio_cursor_init(data, length);
998
		break;
999
#endif /* CONFIG_BLOCK */
1000
	case CEPH_MSG_DATA_NONE:
1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022
	default:
		/* BUG(); */
		break;
	}
}

/*
 * 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.
 */
static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
					size_t *page_offset,
					size_t *length,
					bool *last_piece)
{
	struct page *page;

	switch (data->type) {
	case CEPH_MSG_DATA_PAGELIST:
		page = ceph_msg_data_pagelist_next(data, page_offset, length);
		break;
1023 1024 1025
	case CEPH_MSG_DATA_PAGES:
		page = ceph_msg_data_pages_next(data, page_offset, length);
		break;
1026 1027
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
1028 1029
		page = ceph_msg_data_bio_next(data, page_offset, length);
		break;
1030
#endif /* CONFIG_BLOCK */
1031
	case CEPH_MSG_DATA_NONE:
1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050
	default:
		page = NULL;
		break;
	}
	BUG_ON(!page);
	BUG_ON(*page_offset + *length > PAGE_SIZE);
	BUG_ON(!*length);
	if (last_piece)
		*last_piece = data->cursor.last_piece;

	return page;
}

/*
 * Returns true if the result moves the cursor on to the next piece
 * of the data item.
 */
static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
{
1051
	struct ceph_msg_data_cursor *cursor = &data->cursor;
1052 1053
	bool new_piece;

1054
	BUG_ON(bytes > cursor->resid);
1055 1056 1057 1058
	switch (data->type) {
	case CEPH_MSG_DATA_PAGELIST:
		new_piece = ceph_msg_data_pagelist_advance(data, bytes);
		break;
1059 1060 1061
	case CEPH_MSG_DATA_PAGES:
		new_piece = ceph_msg_data_pages_advance(data, bytes);
		break;
1062 1063
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
1064 1065
		new_piece = ceph_msg_data_bio_advance(data, bytes);
		break;
1066
#endif /* CONFIG_BLOCK */
1067
	case CEPH_MSG_DATA_NONE:
1068 1069 1070 1071 1072 1073 1074 1075
	default:
		BUG();
		break;
	}

	return new_piece;
}

1076 1077
static void prepare_message_data(struct ceph_msg *msg,
				struct ceph_msg_pos *msg_pos)
1078
{
1079 1080
	size_t data_len;

1081
	BUG_ON(!msg);
1082 1083 1084

	data_len = le32_to_cpu(msg->hdr.data_len);
	BUG_ON(!data_len);
1085 1086

	/* initialize page iterator */
1087
	msg_pos->page = 0;
1088
	if (ceph_msg_has_pages(msg))
1089
		msg_pos->page_pos = msg->p.alignment;
1090
	else
1091 1092
		msg_pos->page_pos = 0;
	msg_pos->data_pos = 0;
1093

1094
	/* Initialize data cursors */
1095

1096 1097
#ifdef CONFIG_BLOCK
	if (ceph_msg_has_bio(msg))
1098
		ceph_msg_data_cursor_init(&msg->b, data_len);
1099
#endif /* CONFIG_BLOCK */
1100
	if (ceph_msg_has_pages(msg))
1101
		ceph_msg_data_cursor_init(&msg->p, data_len);
1102
	if (ceph_msg_has_pagelist(msg))
1103
		ceph_msg_data_cursor_init(&msg->l, data_len);
1104

1105
	msg_pos->did_page_crc = false;
1106 1107
}

S
Sage Weil 已提交
1108 1109 1110 1111
/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
1112
static void prepare_write_message_footer(struct ceph_connection *con)
S
Sage Weil 已提交
1113 1114
{
	struct ceph_msg *m = con->out_msg;
1115
	int v = con->out_kvec_left;
S
Sage Weil 已提交
1116

1117 1118
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

S
Sage Weil 已提交
1119 1120 1121 1122 1123 1124 1125
	dout("prepare_write_message_footer %p\n", con);
	con->out_kvec_is_msg = true;
	con->out_kvec[v].iov_base = &m->footer;
	con->out_kvec[v].iov_len = sizeof(m->footer);
	con->out_kvec_bytes += sizeof(m->footer);
	con->out_kvec_left++;
	con->out_more = m->more_to_follow;
1126
	con->out_msg_done = true;
S
Sage Weil 已提交
1127 1128 1129 1130 1131 1132 1133 1134
}

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

1137
	con_out_kvec_reset(con);
S
Sage Weil 已提交
1138
	con->out_kvec_is_msg = true;
1139
	con->out_msg_done = false;
S
Sage Weil 已提交
1140 1141 1142 1143 1144

	/* 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;
1145
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
S
Sage Weil 已提交
1146
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1147
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1148
			&con->out_temp_ack);
S
Sage Weil 已提交
1149 1150
	}

1151
	BUG_ON(list_empty(&con->out_queue));
1152
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
1153
	con->out_msg = m;
1154
	BUG_ON(m->con != con);
1155 1156 1157 1158

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

1160 1161 1162 1163 1164 1165 1166 1167
	/*
	 * 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;
	}
S
Sage Weil 已提交
1168

1169
	dout("prepare_write_message %p seq %lld type %d len %d+%d+%d (%zd)\n",
S
Sage Weil 已提交
1170 1171
	     m, con->out_seq, le16_to_cpu(m->hdr.type),
	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
1172
	     le32_to_cpu(m->hdr.data_len), m->p.length);
S
Sage Weil 已提交
1173 1174 1175
	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

	/* tag + hdr + front + middle */
1176 1177 1178
	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);
1179

S
Sage Weil 已提交
1180
	if (m->middle)
1181
		con_out_kvec_add(con, m->middle->vec.iov_len,
1182
			m->middle->vec.iov_base);
S
Sage Weil 已提交
1183 1184

	/* fill in crc (except data pages), footer */
1185 1186
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
1187
	con->out_msg->footer.flags = 0;
1188 1189 1190 1191 1192 1193 1194 1195

	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 已提交
1196
		con->out_msg->footer.middle_crc = 0;
1197
	dout("%s front_crc %u middle_crc %u\n", __func__,
S
Sage Weil 已提交
1198 1199 1200 1201
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));

	/* is there a data payload? */
1202
	con->out_msg->footer.data_crc = 0;
1203 1204 1205 1206
	if (m->hdr.data_len) {
		prepare_message_data(con->out_msg, &con->out_msg_pos);
		con->out_more = 1;  /* data + footer will follow */
	} else {
S
Sage Weil 已提交
1207
		/* no, queue up footer too and be done */
1208
		prepare_write_message_footer(con);
1209
	}
S
Sage Weil 已提交
1210

1211
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222
}

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

1223
	con_out_kvec_reset(con);
1224

1225
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1226

S
Sage Weil 已提交
1227
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1228
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1229 1230
				&con->out_temp_ack);

S
Sage Weil 已提交
1231
	con->out_more = 1;  /* more will follow.. eventually.. */
1232
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1233 1234
}

1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252
/*
 * 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 已提交
1253 1254 1255 1256 1257 1258
/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
1259 1260
	con_out_kvec_reset(con);
	con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
1261
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1262 1263 1264 1265 1266 1267
}

/*
 * Connection negotiation.
 */

1268 1269
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
1270
{
1271
	struct ceph_auth_handshake *auth;
1272 1273 1274 1275

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;
1276
		return NULL;
1277 1278 1279
	}

	/* Can't hold the mutex while getting authorizer */
1280
	mutex_unlock(&con->mutex);
1281
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1282
	mutex_lock(&con->mutex);
1283

1284
	if (IS_ERR(auth))
1285
		return auth;
1286
	if (con->state != CON_STATE_NEGOTIATING)
1287
		return ERR_PTR(-EAGAIN);
1288

1289 1290
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1291
	return auth;
1292 1293
}

S
Sage Weil 已提交
1294 1295 1296
/*
 * We connected to a peer and are saying hello.
 */
1297
static void prepare_write_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1298
{
1299 1300
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1301
					&con->msgr->my_enc_addr);
1302 1303

	con->out_more = 0;
1304
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1305 1306
}

1307
static int prepare_write_connect(struct ceph_connection *con)
1308
{
1309
	unsigned int global_seq = get_global_seq(con->msgr, 0);
S
Sage Weil 已提交
1310
	int proto;
1311
	int auth_proto;
1312
	struct ceph_auth_handshake *auth;
S
Sage Weil 已提交
1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329

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

1331
	con->out_connect.features = cpu_to_le64(con->msgr->supported_features);
S
Sage Weil 已提交
1332 1333 1334 1335 1336 1337
	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;

1338 1339
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
1340 1341
	if (IS_ERR(auth))
		return PTR_ERR(auth);
1342

1343
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1344 1345 1346
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

1347
	con_out_kvec_add(con, sizeof (con->out_connect),
1348 1349
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
1350
		con_out_kvec_add(con, auth->authorizer_buf_len,
1351
					auth->authorizer_buf);
1352

S
Sage Weil 已提交
1353
	con->out_more = 0;
1354
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1355

1356
	return 0;
S
Sage Weil 已提交
1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378
}

/*
 * 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 */
1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390

		/* 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 已提交
1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401
		}
	}
	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! */
}

1402
static void out_msg_pos_next(struct ceph_connection *con, struct page *page,
A
Alex Elder 已提交
1403
			size_t len, size_t sent)
1404
{
1405
	struct ceph_msg *msg = con->out_msg;
1406
	struct ceph_msg_pos *msg_pos = &con->out_msg_pos;
1407
	bool need_crc = false;
1408

1409 1410
	BUG_ON(!msg);
	BUG_ON(!sent);
1411

1412 1413
	msg_pos->data_pos += sent;
	msg_pos->page_pos += sent;
A
Alex Elder 已提交
1414
	if (ceph_msg_has_pages(msg))
1415
		need_crc = ceph_msg_data_advance(&msg->p, sent);
1416 1417
	else if (ceph_msg_has_pagelist(msg))
		need_crc = ceph_msg_data_advance(&msg->l, sent);
1418 1419 1420 1421
#ifdef CONFIG_BLOCK
	else if (ceph_msg_has_bio(msg))
		need_crc = ceph_msg_data_advance(&msg->b, sent);
#endif /* CONFIG_BLOCK */
1422 1423
	BUG_ON(need_crc && sent != len);

1424 1425
	if (sent < len)
		return;
1426

1427
	BUG_ON(sent != len);
1428 1429 1430
	msg_pos->page_pos = 0;
	msg_pos->page++;
	msg_pos->did_page_crc = false;
1431
}
1432

1433 1434 1435 1436
static void in_msg_pos_next(struct ceph_connection *con, size_t len,
				size_t received)
{
	struct ceph_msg *msg = con->in_msg;
1437
	struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
1438 1439 1440 1441

	BUG_ON(!msg);
	BUG_ON(!received);

1442 1443
	msg_pos->data_pos += received;
	msg_pos->page_pos += received;
1444 1445
	if (ceph_msg_has_pages(msg))
		(void) ceph_msg_data_advance(&msg->p, received);
A
Alex Elder 已提交
1446
#ifdef CONFIG_BLOCK
1447
	else if (ceph_msg_has_bio(msg))
A
Alex Elder 已提交
1448 1449
		(void) ceph_msg_data_advance(&msg->b, received);
#endif /* CONFIG_BLOCK */
1450 1451 1452 1453
	if (received < len)
		return;

	BUG_ON(received != len);
1454 1455
	msg_pos->page_pos = 0;
	msg_pos->page++;
1456 1457
}

1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470
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 已提交
1471 1472 1473 1474 1475 1476 1477
/*
 * 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
 */
1478
static int write_partial_message_data(struct ceph_connection *con)
S
Sage Weil 已提交
1479 1480
{
	struct ceph_msg *msg = con->out_msg;
1481
	struct ceph_msg_pos *msg_pos = &con->out_msg_pos;
1482
	unsigned int data_len = le32_to_cpu(msg->hdr.data_len);
1483
	bool do_datacrc = !con->msgr->nocrc;
S
Sage Weil 已提交
1484 1485
	int ret;

1486
	dout("%s %p msg %p page %d offset %d\n", __func__,
1487
	     con, msg, msg_pos->page, msg_pos->page_pos);
S
Sage Weil 已提交
1488

1489 1490 1491 1492 1493 1494 1495 1496
	/*
	 * 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.
	 */
1497
	while (data_len > msg_pos->data_pos) {
1498
		struct page *page;
1499 1500
		size_t page_offset;
		size_t length;
1501
		bool last_piece;
1502

A
Alex Elder 已提交
1503
		if (ceph_msg_has_pages(msg)) {
1504 1505
			page = ceph_msg_data_next(&msg->p, &page_offset,
							&length, &last_piece);
1506
		} else if (ceph_msg_has_pagelist(msg)) {
1507 1508
			page = ceph_msg_data_next(&msg->l, &page_offset,
							&length, &last_piece);
1509
#ifdef CONFIG_BLOCK
1510
		} else if (ceph_msg_has_bio(msg)) {
1511 1512
			page = ceph_msg_data_next(&msg->b, &page_offset,
							&length, &last_piece);
1513
#endif
S
Sage Weil 已提交
1514
		} else {
A
Alex Elder 已提交
1515 1516 1517 1518
			WARN(1, "con %p data_len %u but no outbound data\n",
				con, data_len);
			ret = -EINVAL;
			goto out;
1519
		}
1520
		if (do_datacrc && !msg_pos->did_page_crc) {
1521
			u32 crc = le32_to_cpu(msg->footer.data_crc);
S
Sage Weil 已提交
1522

1523
			crc = ceph_crc32c_page(crc, page, page_offset, length);
1524
			msg->footer.data_crc = cpu_to_le32(crc);
1525
			msg_pos->did_page_crc = true;
S
Sage Weil 已提交
1526
		}
1527
		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1528
				      length, last_piece);
S
Sage Weil 已提交
1529 1530 1531
		if (ret <= 0)
			goto out;

A
Alex Elder 已提交
1532
		out_msg_pos_next(con, page, length, (size_t) ret);
S
Sage Weil 已提交
1533 1534
	}

1535
	dout("%s %p msg %p done\n", __func__, con, msg);
S
Sage Weil 已提交
1536 1537

	/* prepare and queue up footer, too */
1538
	if (!do_datacrc)
1539
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1540
	con_out_kvec_reset(con);
1541
	prepare_write_message_footer(con);
S
Sage Weil 已提交
1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554
	ret = 1;
out:
	return ret;
}

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

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

1557
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
S
Sage Weil 已提交
1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1570 1571 1572 1573 1574 1575
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587
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;
}

1588 1589 1590 1591 1592 1593 1594
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 已提交
1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615
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,
1616
			int end, int size, void *object)
S
Sage Weil 已提交
1617
{
1618 1619
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
S
Sage Weil 已提交
1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632
		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
 */
1633
static int read_partial_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1634
{
1635 1636 1637
	int size;
	int end;
	int ret;
S
Sage Weil 已提交
1638

1639
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
S
Sage Weil 已提交
1640 1641

	/* peer's banner */
1642 1643 1644
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
S
Sage Weil 已提交
1645 1646
	if (ret <= 0)
		goto out;
1647 1648 1649 1650

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
S
Sage Weil 已提交
1651 1652
	if (ret <= 0)
		goto out;
1653 1654 1655 1656

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

1660 1661 1662 1663 1664 1665
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1666 1667 1668
	int size;
	int end;
	int ret;
1669 1670 1671

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

1672 1673 1674
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
S
Sage Weil 已提交
1675 1676
	if (ret <= 0)
		goto out;
1677 1678 1679 1680

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

1684 1685 1686
	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 已提交
1687 1688 1689
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1690

S
Sage Weil 已提交
1691 1692 1693 1694 1695 1696 1697 1698
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1699
		pr_err("connect to %s got bad banner\n",
1700
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725
		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 已提交
1726
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
S
Sage Weil 已提交
1727
	case AF_INET6:
S
Sage Weil 已提交
1728
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
S
Sage Weil 已提交
1729 1730 1731 1732 1733 1734 1735 1736 1737
	}
	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);
1738
		break;
S
Sage Weil 已提交
1739 1740
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1741
		break;
S
Sage Weil 已提交
1742 1743 1744
	}
}

N
Noah Watkins 已提交
1745 1746 1747 1748 1749 1750
/*
 * 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 已提交
1751 1752
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
N
Noah Watkins 已提交
1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 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

	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 已提交
1840 1841 1842 1843 1844 1845 1846 1847
/*
 * 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 已提交
1848
	int i, ret = -EINVAL;
S
Sage Weil 已提交
1849 1850 1851 1852 1853 1854 1855
	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 已提交
1856 1857 1858 1859 1860 1861
		char delim = ',';

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

N
Noah Watkins 已提交
1863 1864
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
S
Sage Weil 已提交
1865
			goto bad;
N
Noah Watkins 已提交
1866 1867
		ret = -EINVAL;

S
Sage Weil 已提交
1868 1869
		p = ipend;

S
Sage Weil 已提交
1870 1871 1872 1873 1874 1875 1876 1877
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

S
Sage Weil 已提交
1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893
		/* port? */
		if (p < end && *p == ':') {
			port = 0;
			p++;
			while (p < end && *p >= '0' && *p <= '9') {
				port = (port * 10) + (*p - '0');
				p++;
			}
			if (port > 65535 || port == 0)
				goto bad;
		} else {
			port = CEPH_MON_PORT;
		}

		addr_set_port(ss, port);

1894
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
S
Sage Weil 已提交
1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910

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

1916
static int process_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1917
{
1918
	dout("process_banner on %p\n", con);
S
Sage Weil 已提交
1919 1920 1921 1922

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

1923 1924 1925
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

S
Sage Weil 已提交
1926 1927 1928 1929 1930
	/*
	 * 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 已提交
1931 1932
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
S
Sage Weil 已提交
1933 1934
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
Y
Yehuda Sadeh 已提交
1935
		pr_warning("wrong peer, want %s/%d, got %s/%d\n",
1936
			   ceph_pr_addr(&con->peer_addr.in_addr),
Y
Yehuda Sadeh 已提交
1937
			   (int)le32_to_cpu(con->peer_addr.nonce),
1938
			   ceph_pr_addr(&con->actual_peer_addr.in_addr),
Y
Yehuda Sadeh 已提交
1939
			   (int)le32_to_cpu(con->actual_peer_addr.nonce));
1940
		con->error_msg = "wrong peer at address";
S
Sage Weil 已提交
1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953
		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);
1954
		encode_my_addr(con->msgr);
1955
		dout("process_banner learned my addr is %s\n",
1956
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
S
Sage Weil 已提交
1957 1958
	}

1959 1960 1961 1962 1963
	return 0;
}

static int process_connect(struct ceph_connection *con)
{
1964 1965
	u64 sup_feat = con->msgr->supported_features;
	u64 req_feat = con->msgr->required_features;
1966
	u64 server_feat = le64_to_cpu(con->in_reply.features);
1967
	int ret;
1968

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

S
Sage Weil 已提交
1971
	switch (con->in_reply.tag) {
1972 1973 1974 1975
	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),
1976
		       ceph_pr_addr(&con->peer_addr.in_addr),
1977 1978
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
1979
		reset_connection(con);
1980 1981
		return -1;

S
Sage Weil 已提交
1982 1983 1984 1985
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
1986
		       ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
1987 1988 1989
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
1990
		reset_connection(con);
S
Sage Weil 已提交
1991 1992
		return -1;

1993 1994 1995 1996 1997 1998 1999 2000
	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;
		}
2001
		con_out_kvec_reset(con);
2002
		ret = prepare_write_connect(con);
2003 2004
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2005
		prepare_read_connect(con);
2006
		break;
S
Sage Weil 已提交
2007 2008 2009 2010 2011 2012 2013 2014 2015 2016

	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 已提交
2017
		     le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2018 2019
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
2020
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2021
		reset_connection(con);
2022
		con_out_kvec_reset(con);
2023 2024 2025
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2026 2027 2028
		prepare_read_connect(con);

		/* Tell ceph about it. */
2029
		mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2030 2031 2032
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
2033
		mutex_lock(&con->mutex);
2034
		if (con->state != CON_STATE_NEGOTIATING)
2035
			return -EAGAIN;
S
Sage Weil 已提交
2036 2037 2038 2039 2040 2041 2042
		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 已提交
2043
		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
S
Sage Weil 已提交
2044
		     le32_to_cpu(con->out_connect.connect_seq),
S
Sage Weil 已提交
2045 2046
		     le32_to_cpu(con->in_reply.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2047
		con_out_kvec_reset(con);
2048 2049 2050
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2051 2052 2053 2054 2055 2056 2057 2058
		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.
		 */
2059
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
S
Sage Weil 已提交
2060
		     con->peer_global_seq,
S
Sage Weil 已提交
2061
		     le32_to_cpu(con->in_reply.global_seq));
S
Sage Weil 已提交
2062
		get_global_seq(con->msgr,
S
Sage Weil 已提交
2063
			       le32_to_cpu(con->in_reply.global_seq));
2064
		con_out_kvec_reset(con);
2065 2066 2067
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2068 2069 2070
		prepare_read_connect(con);
		break;

2071
	case CEPH_MSGR_TAG_SEQ:
S
Sage Weil 已提交
2072
	case CEPH_MSGR_TAG_READY:
2073 2074 2075 2076
		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),
2077
			       ceph_pr_addr(&con->peer_addr.in_addr),
2078 2079
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
2080
			reset_connection(con);
2081 2082
			return -1;
		}
2083

2084
		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2085
		con->state = CON_STATE_OPEN;
2086
		con->auth_retry = 0;    /* we authenticated; clear flag */
S
Sage Weil 已提交
2087 2088
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
2089
		con->peer_features = server_feat;
S
Sage Weil 已提交
2090 2091 2092 2093 2094 2095
		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 已提交
2096 2097

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

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

2102 2103 2104 2105 2106 2107
		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 已提交
2108 2109 2110 2111 2112 2113 2114 2115 2116
		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 已提交
2117 2118 2119
		pr_err("process_connect got WAIT as client\n");
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
S
Sage Weil 已提交
2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134

	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)
{
2135 2136
	int size = sizeof (con->in_temp_ack);
	int end = size;
S
Sage Weil 已提交
2137

2138
	return read_partial(con, end, size, &con->in_temp_ack);
S
Sage Weil 已提交
2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157
}

/*
 * 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);
2158
		m->ack_stamp = jiffies;
S
Sage Weil 已提交
2159 2160 2161 2162 2163 2164
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}


2165
static int read_partial_message_section(struct ceph_connection *con,
S
Sage Weil 已提交
2166 2167
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
2168
{
2169
	int ret, left;
2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181

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

2185 2186
	return 1;
}
S
Sage Weil 已提交
2187

2188 2189 2190 2191 2192 2193
static int read_partial_msg_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->in_msg;
	struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
	const bool do_datacrc = !con->msgr->nocrc;
	unsigned int data_len;
A
Alex Elder 已提交
2194 2195 2196
	struct page *page;
	size_t page_offset;
	size_t length;
2197 2198 2199 2200 2201 2202
	int ret;

	BUG_ON(!msg);

	data_len = le32_to_cpu(con->in_hdr.data_len);
	while (msg_pos->data_pos < data_len) {
2203
		if (ceph_msg_has_pages(msg)) {
A
Alex Elder 已提交
2204 2205
			page = ceph_msg_data_next(&msg->p, &page_offset,
					&length, NULL);
2206
#ifdef CONFIG_BLOCK
2207
		} else if (ceph_msg_has_bio(msg)) {
A
Alex Elder 已提交
2208 2209
			page = ceph_msg_data_next(&msg->b, &page_offset,
					&length, NULL);
2210 2211 2212 2213
#endif
		} else {
			BUG_ON(1);
		}
A
Alex Elder 已提交
2214 2215 2216 2217 2218 2219 2220 2221 2222
		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
		if (ret <= 0)
			return ret;

		if (do_datacrc)
			con->in_data_crc = ceph_crc32c_page(con->in_data_crc,
							page, page_offset, ret);

		in_msg_pos_next(con, length, ret);
2223 2224 2225 2226 2227
	}

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

S
Sage Weil 已提交
2228 2229 2230
/*
 * read (part of) a message.
 */
A
Alex Elder 已提交
2231 2232
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);

S
Sage Weil 已提交
2233 2234 2235
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
2236 2237
	int size;
	int end;
S
Sage Weil 已提交
2238
	int ret;
2239
	unsigned int front_len, middle_len, data_len;
2240
	bool do_datacrc = !con->msgr->nocrc;
2241
	u64 seq;
2242
	u32 crc;
S
Sage Weil 已提交
2243 2244 2245 2246

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

	/* header */
2247 2248 2249
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
2250 2251
	if (ret <= 0)
		return ret;
2252 2253 2254 2255 2256 2257 2258 2259 2260

	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 已提交
2261 2262 2263 2264
	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);
2265
	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
S
Sage Weil 已提交
2266 2267 2268 2269 2270
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

2271 2272 2273
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
S
Sage Weil 已提交
2274
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2275
			ENTITY_NAME(con->peer_name),
2276
			ceph_pr_addr(&con->peer_addr.in_addr),
2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288
			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 已提交
2289 2290
	/* allocate message? */
	if (!con->in_msg) {
2291 2292
		int skip = 0;

S
Sage Weil 已提交
2293
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2294
		     front_len, data_len);
2295 2296 2297
		ret = ceph_con_in_msg_alloc(con, &skip);
		if (ret < 0)
			return ret;
2298
		if (skip) {
S
Sage Weil 已提交
2299
			/* skip this message */
2300
			dout("alloc_msg said skip message\n");
2301
			BUG_ON(con->in_msg);
S
Sage Weil 已提交
2302 2303 2304
			con->in_base_pos = -front_len - middle_len - data_len -
				sizeof(m->footer);
			con->in_tag = CEPH_MSGR_TAG_READY;
2305
			con->in_seq++;
S
Sage Weil 已提交
2306 2307
			return 0;
		}
2308

2309
		BUG_ON(!con->in_msg);
2310
		BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
2311 2312
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
2313 2314
		if (m->middle)
			m->middle->vec.iov_len = 0;
2315

2316
		/* prepare for data payload, if any */
2317

2318 2319
		if (data_len)
			prepare_message_data(con->in_msg, &con->in_msg_pos);
S
Sage Weil 已提交
2320 2321 2322
	}

	/* front */
2323 2324 2325 2326
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
S
Sage Weil 已提交
2327 2328

	/* middle */
2329
	if (m->middle) {
S
Sage Weil 已提交
2330 2331
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
2332
						   &con->in_middle_crc);
S
Sage Weil 已提交
2333 2334 2335 2336 2337
		if (ret <= 0)
			return ret;
	}

	/* (page) data */
2338 2339 2340 2341
	if (data_len) {
		ret = read_partial_msg_data(con);
		if (ret <= 0)
			return ret;
S
Sage Weil 已提交
2342 2343 2344
	}

	/* footer */
2345 2346 2347
	size = sizeof (m->footer);
	end += size;
	ret = read_partial(con, end, size, &m->footer);
2348 2349 2350
	if (ret <= 0)
		return ret;

S
Sage Weil 已提交
2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365
	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;
	}
2366
	if (do_datacrc &&
S
Sage Weil 已提交
2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383
	    (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;
	}

	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 已提交
2384
	struct ceph_msg *msg;
S
Sage Weil 已提交
2385

2386 2387
	BUG_ON(con->in_msg->con != con);
	con->in_msg->con = NULL;
S
Sage Weil 已提交
2388
	msg = con->in_msg;
S
Sage Weil 已提交
2389
	con->in_msg = NULL;
S
Sage Weil 已提交
2390
	con->ops->put(con);
S
Sage Weil 已提交
2391 2392 2393

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

	con->in_seq++;
2397
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2398 2399 2400

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
2401
	     ENTITY_NAME(msg->hdr.src),
S
Sage Weil 已提交
2402 2403 2404 2405 2406 2407
	     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);
2408 2409

	mutex_lock(&con->mutex);
S
Sage Weil 已提交
2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420
}


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

2421
	dout("try_write start %p state %lu\n", con, con->state);
S
Sage Weil 已提交
2422 2423 2424 2425 2426

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

	/* open the socket first? */
2427 2428 2429
	if (con->state == CON_STATE_PREOPEN) {
		BUG_ON(con->sock);
		con->state = CON_STATE_CONNECTING;
2430

2431
		con_out_kvec_reset(con);
2432
		prepare_write_banner(con);
2433
		prepare_read_banner(con);
S
Sage Weil 已提交
2434

2435
		BUG_ON(con->in_msg);
S
Sage Weil 已提交
2436 2437 2438
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
2439 2440
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
S
Sage Weil 已提交
2441 2442 2443 2444 2445 2446 2447 2448 2449 2450
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
	if (con->out_skip) {
		ret = write_partial_skip(con);
		if (ret <= 0)
2451
			goto out;
S
Sage Weil 已提交
2452 2453 2454 2455
	}
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
		if (ret <= 0)
2456
			goto out;
S
Sage Weil 已提交
2457 2458 2459 2460
	}

	/* msg pages? */
	if (con->out_msg) {
2461 2462 2463 2464 2465 2466
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

2467
		ret = write_partial_message_data(con);
S
Sage Weil 已提交
2468 2469 2470
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2471
			goto out;
S
Sage Weil 已提交
2472
		if (ret < 0) {
2473
			dout("try_write write_partial_message_data err %d\n",
S
Sage Weil 已提交
2474
			     ret);
2475
			goto out;
S
Sage Weil 已提交
2476 2477 2478
		}
	}

2479
do_next:
2480
	if (con->state == CON_STATE_OPEN) {
S
Sage Weil 已提交
2481 2482 2483 2484 2485 2486 2487 2488 2489
		/* 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;
		}
2490
		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
2491 2492 2493 2494 2495 2496
			prepare_write_keepalive(con);
			goto more;
		}
	}

	/* Nothing to do! */
2497
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
2498 2499 2500
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2501
	dout("try_write done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513
	return ret;
}



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

2514 2515 2516 2517 2518
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 已提交
2519 2520
		return 0;

2521
	BUG_ON(!con->sock);
2522

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

2526
	if (con->state == CON_STATE_CONNECTING) {
2527 2528 2529
		dout("try_read connecting\n");
		ret = read_partial_banner(con);
		if (ret <= 0)
2530
			goto out;
2531 2532 2533 2534
		ret = process_banner(con);
		if (ret < 0)
			goto out;

2535
		con->state = CON_STATE_NEGOTIATING;
2536

2537 2538 2539 2540 2541
		/*
		 * Received banner is good, exchange connection info.
		 * Do not reset out_kvec, as sending our banner raced
		 * with receiving peer banner after connect completed.
		 */
2542 2543 2544 2545 2546 2547
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
		prepare_read_connect(con);

		/* Send connection info before awaiting response */
2548 2549 2550
		goto out;
	}

2551
	if (con->state == CON_STATE_NEGOTIATING) {
2552
		dout("try_read negotiating\n");
S
Sage Weil 已提交
2553 2554 2555
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2556 2557 2558
		ret = process_connect(con);
		if (ret < 0)
			goto out;
S
Sage Weil 已提交
2559 2560 2561
		goto more;
	}

2562
	WARN_ON(con->state != CON_STATE_OPEN);
2563

S
Sage Weil 已提交
2564 2565 2566 2567 2568 2569
	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
A
Alex Elder 已提交
2570 2571 2572
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

S
Sage Weil 已提交
2573 2574 2575
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2576
			goto out;
S
Sage Weil 已提交
2577 2578 2579 2580 2581 2582 2583 2584 2585 2586
		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)
2587
			goto out;
S
Sage Weil 已提交
2588 2589 2590 2591 2592 2593 2594 2595 2596
		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:
2597 2598
			con_close_socket(con);
			con->state = CON_STATE_CLOSED;
2599
			goto out;
S
Sage Weil 已提交
2600 2601 2602 2603 2604 2605 2606 2607 2608 2609 2610
		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;
2611
				break;
S
Sage Weil 已提交
2612 2613
			case -EIO:
				con->error_msg = "io error";
2614
				break;
S
Sage Weil 已提交
2615
			}
2616
			goto out;
S
Sage Weil 已提交
2617 2618 2619 2620
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
2621 2622
		if (con->state == CON_STATE_OPEN)
			prepare_read_tag(con);
S
Sage Weil 已提交
2623 2624
		goto more;
	}
2625 2626 2627 2628 2629 2630
	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 已提交
2631 2632
		ret = read_partial_ack(con);
		if (ret <= 0)
2633
			goto out;
S
Sage Weil 已提交
2634 2635 2636 2637 2638
		process_ack(con);
		goto more;
	}

out:
2639
	dout("try_read done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650
	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 已提交
2651 2652 2653
 * 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 已提交
2654
 */
A
Alex Elder 已提交
2655
static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
S
Sage Weil 已提交
2656 2657
{
	if (!con->ops->get(con)) {
A
Alex Elder 已提交
2658 2659 2660
		dout("%s %p ref count 0\n", __func__, con);

		return -ENOENT;
S
Sage Weil 已提交
2661 2662
	}

A
Alex Elder 已提交
2663 2664
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
		dout("%s %p - already queued\n", __func__, con);
S
Sage Weil 已提交
2665
		con->ops->put(con);
A
Alex Elder 已提交
2666 2667

		return -EBUSY;
S
Sage Weil 已提交
2668
	}
A
Alex Elder 已提交
2669 2670 2671 2672 2673 2674 2675 2676 2677

	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 已提交
2678 2679
}

2680 2681
static bool con_sock_closed(struct ceph_connection *con)
{
2682
	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707 2708
		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:
		pr_warning("%s con %p unrecognized state %lu\n",
			__func__, con, con->state);
		con->error_msg = "unrecognized con state";
		BUG();
		break;
	}
#undef CASE

	return true;
}

2709 2710 2711 2712 2713 2714 2715 2716 2717 2718 2719 2720 2721 2722 2723 2724 2725 2726
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;
}

2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743
/* 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 已提交
2744 2745 2746 2747 2748 2749 2750
/*
 * 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);
2751
	bool fault;
S
Sage Weil 已提交
2752

S
Sage Weil 已提交
2753
	mutex_lock(&con->mutex);
2754 2755
	while (true) {
		int ret;
S
Sage Weil 已提交
2756

2757 2758 2759 2760 2761 2762 2763 2764 2765 2766 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777
		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);
		}
2778

2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796
		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 */
2797
	}
2798 2799
	if (fault)
		con_fault(con);
S
Sage Weil 已提交
2800
	mutex_unlock(&con->mutex);
2801

2802 2803 2804 2805
	if (fault)
		con_fault_finish(con);

	con->ops->put(con);
S
Sage Weil 已提交
2806 2807 2808 2809 2810 2811
}

/*
 * Generic error/fault handler.  A retry mechanism is used with
 * exponential backoff
 */
2812
static void con_fault(struct ceph_connection *con)
S
Sage Weil 已提交
2813
{
2814
	pr_warning("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2815
	       ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
S
Sage Weil 已提交
2816
	dout("fault %p state %lu to peer %s\n",
2817
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2818

2819
	WARN_ON(con->state != CON_STATE_CONNECTING &&
2820 2821
	       con->state != CON_STATE_NEGOTIATING &&
	       con->state != CON_STATE_OPEN);
2822

S
Sage Weil 已提交
2823
	con_close_socket(con);
S
Sage Weil 已提交
2824

2825
	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2826 2827
		dout("fault on LOSSYTX channel, marking CLOSED\n");
		con->state = CON_STATE_CLOSED;
2828
		return;
2829 2830
	}

S
Sage Weil 已提交
2831
	if (con->in_msg) {
2832 2833
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
S
Sage Weil 已提交
2834 2835
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
S
Sage Weil 已提交
2836
		con->ops->put(con);
S
Sage Weil 已提交
2837
	}
S
Sage Weil 已提交
2838

2839 2840
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2841

S
Sage Weil 已提交
2842 2843 2844
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2845
	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
2846
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2847
		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2848
		con->state = CON_STATE_STANDBY;
2849 2850
	} else {
		/* retry after a delay. */
2851
		con->state = CON_STATE_PREOPEN;
2852 2853 2854 2855
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
2856
		con_flag_set(con, CON_FLAG_BACKOFF);
A
Alex Elder 已提交
2857
		queue_con(con);
S
Sage Weil 已提交
2858 2859 2860 2861 2862 2863
	}
}



/*
2864
 * initialize a new messenger instance
S
Sage Weil 已提交
2865
 */
2866 2867 2868 2869 2870
void ceph_messenger_init(struct ceph_messenger *msgr,
			struct ceph_entity_addr *myaddr,
			u32 supported_features,
			u32 required_features,
			bool nocrc)
S
Sage Weil 已提交
2871
{
2872 2873 2874
	msgr->supported_features = supported_features;
	msgr->required_features = required_features;

S
Sage Weil 已提交
2875 2876 2877 2878 2879 2880
	spin_lock_init(&msgr->global_seq_lock);

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

	/* select a random nonce */
2881
	msgr->inst.addr.type = 0;
S
Sage Weil 已提交
2882
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2883
	encode_my_addr(msgr);
2884
	msgr->nocrc = nocrc;
S
Sage Weil 已提交
2885

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

2888
	dout("%s %p\n", __func__, msgr);
S
Sage Weil 已提交
2889
}
2890
EXPORT_SYMBOL(ceph_messenger_init);
S
Sage Weil 已提交
2891

S
Sage Weil 已提交
2892 2893 2894
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
2895
	if (con->state == CON_STATE_STANDBY) {
S
Sage Weil 已提交
2896
		dout("clear_standby %p and ++connect_seq\n", con);
2897
		con->state = CON_STATE_PREOPEN;
S
Sage Weil 已提交
2898
		con->connect_seq++;
2899 2900
		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
S
Sage Weil 已提交
2901 2902 2903
	}
}

S
Sage Weil 已提交
2904 2905 2906 2907 2908 2909
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	/* set src+dst */
2910
	msg->hdr.src = con->msgr->inst.name;
2911
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2912 2913
	msg->needs_out_seq = true;

2914
	mutex_lock(&con->mutex);
2915

2916
	if (con->state == CON_STATE_CLOSED) {
2917 2918 2919 2920 2921 2922
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		mutex_unlock(&con->mutex);
		return;
	}

2923
	BUG_ON(msg->con != NULL);
S
Sage Weil 已提交
2924
	msg->con = con->ops->get(con);
2925 2926
	BUG_ON(msg->con == NULL);

S
Sage Weil 已提交
2927 2928 2929 2930 2931 2932 2933 2934
	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));
2935 2936

	clear_standby(con);
2937
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2938 2939 2940

	/* if there wasn't anything waiting to send before, queue
	 * new work */
2941
	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
2942 2943
		queue_con(con);
}
2944
EXPORT_SYMBOL(ceph_con_send);
S
Sage Weil 已提交
2945 2946 2947 2948

/*
 * Revoke a message that was previously queued for send
 */
2949
void ceph_msg_revoke(struct ceph_msg *msg)
S
Sage Weil 已提交
2950
{
2951 2952 2953 2954 2955
	struct ceph_connection *con = msg->con;

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

2956
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
2957
	if (!list_empty(&msg->list_head)) {
2958
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
S
Sage Weil 已提交
2959
		list_del_init(&msg->list_head);
2960
		BUG_ON(msg->con == NULL);
S
Sage Weil 已提交
2961
		msg->con->ops->put(msg->con);
2962
		msg->con = NULL;
S
Sage Weil 已提交
2963
		msg->hdr.seq = 0;
2964

S
Sage Weil 已提交
2965
		ceph_msg_put(msg);
S
Sage Weil 已提交
2966 2967
	}
	if (con->out_msg == msg) {
2968
		dout("%s %p msg %p - was sending\n", __func__, con, msg);
S
Sage Weil 已提交
2969
		con->out_msg = NULL;
S
Sage Weil 已提交
2970 2971 2972 2973
		if (con->out_kvec_is_msg) {
			con->out_skip = con->out_kvec_bytes;
			con->out_kvec_is_msg = false;
		}
S
Sage Weil 已提交
2974
		msg->hdr.seq = 0;
2975 2976

		ceph_msg_put(msg);
S
Sage Weil 已提交
2977
	}
2978
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2979 2980
}

2981
/*
2982
 * Revoke a message that we may be reading data into
2983
 */
2984
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2985
{
2986 2987 2988 2989 2990 2991 2992 2993 2994 2995
	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;
2996
	mutex_lock(&con->mutex);
2997
	if (con->in_msg == msg) {
2998 2999 3000
		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);
3001 3002

		/* skip rest of message */
3003 3004
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
3005
				sizeof(struct ceph_msg_header) -
3006 3007 3008
				front_len -
				middle_len -
				data_len -
3009 3010 3011 3012
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
3013
		con->in_seq++;
3014
	} else {
3015 3016
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
3017 3018 3019 3020
	}
	mutex_unlock(&con->mutex);
}

S
Sage Weil 已提交
3021 3022 3023 3024 3025
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
S
Sage Weil 已提交
3026
	dout("con_keepalive %p\n", con);
3027
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3028
	clear_standby(con);
3029
	mutex_unlock(&con->mutex);
3030 3031
	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 已提交
3032 3033
		queue_con(con);
}
3034
EXPORT_SYMBOL(ceph_con_keepalive);
S
Sage Weil 已提交
3035

A
Alex Elder 已提交
3036 3037 3038 3039 3040
static void ceph_msg_data_init(struct ceph_msg_data *data)
{
	data->type = CEPH_MSG_DATA_NONE;
}

3041
void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
3042
		size_t length, size_t alignment)
3043
{
3044 3045
	BUG_ON(!pages);
	BUG_ON(!length);
A
Alex Elder 已提交
3046
	BUG_ON(msg->p.type != CEPH_MSG_DATA_NONE);
3047

A
Alex Elder 已提交
3048
	msg->p.type = CEPH_MSG_DATA_PAGES;
3049 3050 3051
	msg->p.pages = pages;
	msg->p.length = length;
	msg->p.alignment = alignment & ~PAGE_MASK;
3052 3053
}
EXPORT_SYMBOL(ceph_msg_data_set_pages);
S
Sage Weil 已提交
3054

3055 3056 3057
void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
				struct ceph_pagelist *pagelist)
{
3058 3059
	BUG_ON(!pagelist);
	BUG_ON(!pagelist->length);
A
Alex Elder 已提交
3060
	BUG_ON(msg->l.type != CEPH_MSG_DATA_NONE);
3061

A
Alex Elder 已提交
3062
	msg->l.type = CEPH_MSG_DATA_PAGELIST;
3063
	msg->l.pagelist = pagelist;
3064 3065 3066 3067 3068
}
EXPORT_SYMBOL(ceph_msg_data_set_pagelist);

void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
{
3069
	BUG_ON(!bio);
A
Alex Elder 已提交
3070
	BUG_ON(msg->b.type != CEPH_MSG_DATA_NONE);
3071

A
Alex Elder 已提交
3072
	msg->b.type = CEPH_MSG_DATA_BIO;
3073
	msg->b.bio = bio;
3074 3075 3076
}
EXPORT_SYMBOL(ceph_msg_data_set_bio);

S
Sage Weil 已提交
3077 3078 3079 3080
/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
3081 3082
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
S
Sage Weil 已提交
3083 3084 3085
{
	struct ceph_msg *m;

3086
	m = kzalloc(sizeof(*m), flags);
S
Sage Weil 已提交
3087 3088 3089 3090
	if (m == NULL)
		goto out;

	m->hdr.type = cpu_to_le16(type);
3091
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
S
Sage Weil 已提交
3092
	m->hdr.front_len = cpu_to_le32(front_len);
3093

3094 3095
	INIT_LIST_HEAD(&m->list_head);
	kref_init(&m->kref);
3096

A
Alex Elder 已提交
3097 3098 3099 3100
	ceph_msg_data_init(&m->p);
	ceph_msg_data_init(&m->l);
	ceph_msg_data_init(&m->b);

S
Sage Weil 已提交
3101
	/* front */
3102
	m->front_max = front_len;
S
Sage Weil 已提交
3103 3104
	if (front_len) {
		if (front_len > PAGE_CACHE_SIZE) {
3105
			m->front.iov_base = __vmalloc(front_len, flags,
S
Sage Weil 已提交
3106 3107 3108
						      PAGE_KERNEL);
			m->front_is_vmalloc = true;
		} else {
3109
			m->front.iov_base = kmalloc(front_len, flags);
S
Sage Weil 已提交
3110 3111
		}
		if (m->front.iov_base == NULL) {
3112
			dout("ceph_msg_new can't allocate %d bytes\n",
S
Sage Weil 已提交
3113 3114 3115 3116 3117 3118 3119 3120
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
	m->front.iov_len = front_len;

S
Sage Weil 已提交
3121
	dout("ceph_msg_new %p front %d\n", m, front_len);
S
Sage Weil 已提交
3122 3123 3124 3125 3126
	return m;

out2:
	ceph_msg_put(m);
out:
3127 3128 3129
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
3130
		WARN_ON(1);
3131 3132 3133 3134
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
3135
	return NULL;
S
Sage Weil 已提交
3136
}
3137
EXPORT_SYMBOL(ceph_msg_new);
S
Sage Weil 已提交
3138 3139 3140 3141 3142 3143 3144 3145

/*
 * 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.
 */
3146
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
S
Sage Weil 已提交
3147 3148 3149 3150 3151 3152 3153 3154 3155
{
	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 已提交
3156
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
S
Sage Weil 已提交
3157 3158 3159 3160 3161
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

3162
/*
A
Alex Elder 已提交
3163 3164 3165 3166
 * 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.
 *
3167 3168 3169 3170 3171 3172 3173 3174 3175
 * 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
3176
 */
3177
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3178
{
3179
	struct ceph_msg_header *hdr = &con->in_hdr;
3180
	int middle_len = le32_to_cpu(hdr->middle_len);
3181
	struct ceph_msg *msg;
3182
	int ret = 0;
3183

A
Alex Elder 已提交
3184
	BUG_ON(con->in_msg != NULL);
3185
	BUG_ON(!con->ops->alloc_msg);
3186

3187 3188 3189 3190 3191
	mutex_unlock(&con->mutex);
	msg = con->ops->alloc_msg(con, hdr, skip);
	mutex_lock(&con->mutex);
	if (con->state != CON_STATE_OPEN) {
		if (msg)
3192
			ceph_msg_put(msg);
3193 3194
		return -EAGAIN;
	}
3195 3196 3197
	if (msg) {
		BUG_ON(*skip);
		con->in_msg = msg;
S
Sage Weil 已提交
3198
		con->in_msg->con = con->ops->get(con);
3199
		BUG_ON(con->in_msg->con == NULL);
3200 3201 3202 3203 3204 3205 3206 3207 3208 3209
	} 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";

3210
		return -ENOMEM;
3211
	}
A
Alex Elder 已提交
3212
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3213

A
Alex Elder 已提交
3214 3215
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
3216
		if (ret < 0) {
A
Alex Elder 已提交
3217 3218
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
3219 3220
		}
	}
3221

3222
	return ret;
3223 3224
}

S
Sage Weil 已提交
3225 3226 3227 3228 3229 3230 3231 3232 3233 3234 3235 3236 3237 3238 3239 3240 3241

/*
 * Free a generically kmalloc'd message.
 */
void ceph_msg_kfree(struct ceph_msg *m)
{
	dout("msg_kfree %p\n", m);
	if (m->front_is_vmalloc)
		vfree(m->front.iov_base);
	else
		kfree(m->front.iov_base);
	kfree(m);
}

/*
 * Drop a msg ref.  Destroy as needed.
 */
S
Sage Weil 已提交
3242 3243 3244
void ceph_msg_last_put(struct kref *kref)
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
S
Sage Weil 已提交
3245

S
Sage Weil 已提交
3246 3247 3248 3249 3250 3251 3252
	dout("ceph_msg_put last one on %p\n", m);
	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 已提交
3253
	}
3254
	if (ceph_msg_has_pages(m)) {
3255 3256
		m->p.length = 0;
		m->p.pages = NULL;
3257
		m->p.type = CEPH_OSD_DATA_TYPE_NONE;
3258 3259
	}
	if (ceph_msg_has_pagelist(m)) {
3260 3261 3262
		ceph_pagelist_release(m->l.pagelist);
		kfree(m->l.pagelist);
		m->l.pagelist = NULL;
3263 3264 3265 3266 3267
		m->l.type = CEPH_OSD_DATA_TYPE_NONE;
	}
	if (ceph_msg_has_bio(m)) {
		m->b.bio = NULL;
		m->b.type = CEPH_OSD_DATA_TYPE_NONE;
3268 3269
	}

S
Sage Weil 已提交
3270 3271 3272 3273
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
		ceph_msg_kfree(m);
S
Sage Weil 已提交
3274
}
3275
EXPORT_SYMBOL(ceph_msg_last_put);
3276 3277 3278

void ceph_msg_dump(struct ceph_msg *msg)
{
3279
	pr_debug("msg_dump %p (front_max %d length %zd)\n", msg,
3280
		 msg->front_max, msg->p.length);
3281 3282 3283 3284 3285 3286 3287 3288 3289 3290 3291 3292 3293 3294 3295
	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);
}
3296
EXPORT_SYMBOL(ceph_msg_dump);