messenger.c 66.8 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 13
#include <linux/bio.h>
#include <linux/blkdev.h>
N
Noah Watkins 已提交
14
#include <linux/dns_resolver.h>
S
Sage Weil 已提交
15 16
#include <net/tcp.h>

17 18 19 20
#include <linux/ceph/libceph.h>
#include <linux/ceph/messenger.h>
#include <linux/ceph/decode.h>
#include <linux/ceph/pagelist.h>
21
#include <linux/export.h>
S
Sage Weil 已提交
22 23 24 25 26 27 28 29 30 31

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

32 33 34 35 36 37 38 39
/* State values for ceph_connection->sock_state; NEW is assumed to be 0 */

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

S
Sage Weil 已提交
40 41 42 43 44
/* 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;

45 46 47 48
#ifdef CONFIG_LOCKDEP
static struct lock_class_key socket_class;
#endif

A
Alex Elder 已提交
49 50 51 52 53
/*
 * 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 已提交
54 55 56 57 58 59

static void queue_con(struct ceph_connection *con);
static void con_work(struct work_struct *);
static void ceph_fault(struct ceph_connection *con);

/*
60 61
 * Nicely render a sockaddr as a string.  An array of formatted
 * strings is used, to approximate reentrancy.
S
Sage Weil 已提交
62
 */
63 64 65 66 67 68 69
#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 已提交
70

71 72
static struct page *zero_page;		/* used in certain error cases */

73
const char *ceph_pr_addr(const struct sockaddr_storage *ss)
S
Sage Weil 已提交
74 75 76
{
	int i;
	char *s;
A
Alex Elder 已提交
77 78
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
S
Sage Weil 已提交
79

80
	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
S
Sage Weil 已提交
81 82 83 84
	s = addr_str[i];

	switch (ss->ss_family) {
	case AF_INET:
A
Alex Elder 已提交
85 86
		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
			 ntohs(in4->sin_port));
S
Sage Weil 已提交
87 88 89
		break;

	case AF_INET6:
A
Alex Elder 已提交
90 91
		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
			 ntohs(in6->sin6_port));
S
Sage Weil 已提交
92 93 94
		break;

	default:
A
Alex Elder 已提交
95 96
		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
			 ss->ss_family);
S
Sage Weil 已提交
97 98 99 100
	}

	return s;
}
101
EXPORT_SYMBOL(ceph_pr_addr);
S
Sage Weil 已提交
102

103 104 105 106 107 108
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 已提交
109 110 111
/*
 * work queue for all reading and writing to/from the socket.
 */
A
Alex Elder 已提交
112
static struct workqueue_struct *ceph_msgr_wq;
S
Sage Weil 已提交
113

114 115
void _ceph_msgr_exit(void)
{
A
Alex Elder 已提交
116
	if (ceph_msgr_wq) {
117
		destroy_workqueue(ceph_msgr_wq);
A
Alex Elder 已提交
118 119
		ceph_msgr_wq = NULL;
	}
120 121 122 123 124 125 126

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

127
int ceph_msgr_init(void)
S
Sage Weil 已提交
128
{
129 130 131 132
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
	page_cache_get(zero_page);

133
	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_NON_REENTRANT, 0);
134 135
	if (ceph_msgr_wq)
		return 0;
136

137 138
	pr_err("msgr_init failed to create workqueue\n");
	_ceph_msgr_exit();
139

140
	return -ENOMEM;
S
Sage Weil 已提交
141
}
142
EXPORT_SYMBOL(ceph_msgr_init);
S
Sage Weil 已提交
143 144 145

void ceph_msgr_exit(void)
{
146 147
	BUG_ON(ceph_msgr_wq == NULL);

148
	_ceph_msgr_exit();
S
Sage Weil 已提交
149
}
150
EXPORT_SYMBOL(ceph_msgr_exit);
S
Sage Weil 已提交
151

Y
Yehuda Sadeh 已提交
152
void ceph_msgr_flush(void)
153 154 155
{
	flush_workqueue(ceph_msgr_wq);
}
156
EXPORT_SYMBOL(ceph_msgr_flush);
157

158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206
/* 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);
}

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

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

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

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 &&
			old_state != CON_SOCK_STATE_CLOSING))
		printk("%s: unexpected old state %d\n", __func__, old_state);
}
207

S
Sage Weil 已提交
208 209 210 211 212
/*
 * socket callback functions
 */

/* data available on socket, or listen socket received a connect */
A
Alex Elder 已提交
213
static void ceph_sock_data_ready(struct sock *sk, int count_unused)
S
Sage Weil 已提交
214
{
A
Alex Elder 已提交
215 216
	struct ceph_connection *con = sk->sk_user_data;

S
Sage Weil 已提交
217
	if (sk->sk_state != TCP_CLOSE_WAIT) {
A
Alex Elder 已提交
218
		dout("%s on %p state = %lu, queueing work\n", __func__,
S
Sage Weil 已提交
219 220 221 222 223 224
		     con, con->state);
		queue_con(con);
	}
}

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

229 230
	/* 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 已提交
231
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
232 233 234 235
	 * 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().
	 */
236
	if (test_bit(WRITE_PENDING, &con->flags)) {
237
		if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
A
Alex Elder 已提交
238
			dout("%s %p queueing write work\n", __func__, con);
239 240 241
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
S
Sage Weil 已提交
242
	} else {
A
Alex Elder 已提交
243
		dout("%s %p nothing to write\n", __func__, con);
S
Sage Weil 已提交
244 245 246 247
	}
}

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

A
Alex Elder 已提交
252
	dout("%s %p state = %lu sk_state = %u\n", __func__,
S
Sage Weil 已提交
253 254 255 256 257 258 259
	     con, con->state, sk->sk_state);

	if (test_bit(CLOSED, &con->state))
		return;

	switch (sk->sk_state) {
	case TCP_CLOSE:
A
Alex Elder 已提交
260
		dout("%s TCP_CLOSE\n", __func__);
S
Sage Weil 已提交
261
	case TCP_CLOSE_WAIT:
A
Alex Elder 已提交
262
		dout("%s TCP_CLOSE_WAIT\n", __func__);
263
		con_sock_state_closing(con);
264
		if (test_and_set_bit(SOCK_CLOSED, &con->flags) == 0) {
S
Sage Weil 已提交
265 266 267 268 269 270 271 272
			if (test_bit(CONNECTING, &con->state))
				con->error_msg = "connection failed";
			else
				con->error_msg = "socket closed";
			queue_con(con);
		}
		break;
	case TCP_ESTABLISHED:
A
Alex Elder 已提交
273
		dout("%s TCP_ESTABLISHED\n", __func__);
274
		con_sock_state_connected(con);
S
Sage Weil 已提交
275 276
		queue_con(con);
		break;
A
Alex Elder 已提交
277 278
	default:	/* Everything else is uninteresting */
		break;
S
Sage Weil 已提交
279 280 281 282 283 284 285 286 287 288
	}
}

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
A
Alex Elder 已提交
289
	sk->sk_user_data = con;
A
Alex Elder 已提交
290 291 292
	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 已提交
293 294 295 296 297 298 299 300 301 302
}


/*
 * socket helpers
 */

/*
 * initiate connection to a remote socket.
 */
303
static int ceph_tcp_connect(struct ceph_connection *con)
S
Sage Weil 已提交
304
{
S
Sage Weil 已提交
305
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
S
Sage Weil 已提交
306 307 308 309
	struct socket *sock;
	int ret;

	BUG_ON(con->sock);
S
Sage Weil 已提交
310 311
	ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
			       IPPROTO_TCP, &sock);
S
Sage Weil 已提交
312
	if (ret)
313
		return ret;
S
Sage Weil 已提交
314 315
	sock->sk->sk_allocation = GFP_NOFS;

316 317 318 319
#ifdef CONFIG_LOCKDEP
	lockdep_set_class(&sock->sk->sk_lock, &socket_class);
#endif

S
Sage Weil 已提交
320 321
	set_sock_callbacks(sock, con);

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

S
Sage Weil 已提交
324 325
	ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
				 O_NONBLOCK);
S
Sage Weil 已提交
326 327
	if (ret == -EINPROGRESS) {
		dout("connect %s EINPROGRESS sk_state = %u\n",
328
		     ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
329
		     sock->sk->sk_state);
330
	} else if (ret < 0) {
S
Sage Weil 已提交
331
		pr_err("connect %s error %d\n",
332
		       ceph_pr_addr(&con->peer_addr.in_addr), ret);
S
Sage Weil 已提交
333 334 335
		sock_release(sock);
		con->error_msg = "connect error";

336
		return ret;
337 338
	}
	con->sock = sock;
339
	con_sock_state_connecting(con);
340

341
	return 0;
S
Sage Weil 已提交
342 343 344 345 346 347
}

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

350 351 352 353
	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
354 355 356 357 358 359 360 361 362 363
}

/*
 * 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 };
364
	int r;
S
Sage Weil 已提交
365 366 367 368 369 370

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

371 372 373 374
	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
375 376
}

377 378 379 380 381 382 383 384 385 386 387 388 389
static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
		     int offset, size_t size, int more)
{
	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 已提交
390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405

/*
 * Shutdown/close the socket for the given connection.
 */
static int con_close_socket(struct ceph_connection *con)
{
	int rc;

	dout("con_close_socket on %p sock %p\n", con, con->sock);
	if (!con->sock)
		return 0;
	set_bit(SOCK_CLOSED, &con->state);
	rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
	sock_release(con->sock);
	con->sock = NULL;
	clear_bit(SOCK_CLOSED, &con->state);
406
	con_sock_state_closed(con);
S
Sage Weil 已提交
407 408 409 410 411 412 413 414 415 416
	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);
417
	BUG_ON(msg->con == NULL);
418
	ceph_con_put(msg->con);
419 420
	msg->con = NULL;

S
Sage Weil 已提交
421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438
	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 */
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

439
	if (con->in_msg) {
440 441
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
442 443
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
444
		ceph_con_put(con->in_msg->con);
445 446
	}

S
Sage Weil 已提交
447 448
	con->connect_seq = 0;
	con->out_seq = 0;
449 450 451 452
	if (con->out_msg) {
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
S
Sage Weil 已提交
453
	con->in_seq = 0;
454
	con->in_seq_acked = 0;
S
Sage Weil 已提交
455 456 457 458 459 460 461
}

/*
 * mark a peer down.  drop any open connections.
 */
void ceph_con_close(struct ceph_connection *con)
{
462 463
	dout("con_close %p peer %s\n", con,
	     ceph_pr_addr(&con->peer_addr.in_addr));
464
	clear_bit(NEGOTIATING, &con->state);
S
Sage Weil 已提交
465
	clear_bit(STANDBY, &con->state);  /* avoid connect_seq bump */
466 467
	set_bit(CLOSED, &con->state);

468 469 470
	clear_bit(LOSSYTX, &con->flags);  /* so we retry next connect */
	clear_bit(KEEPALIVE_PENDING, &con->flags);
	clear_bit(WRITE_PENDING, &con->flags);
471

472
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
473
	reset_connection(con);
S
Sage Weil 已提交
474
	con->peer_global_seq = 0;
475
	cancel_delayed_work(&con->work);
476
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
477 478
	queue_con(con);
}
479
EXPORT_SYMBOL(ceph_con_close);
S
Sage Weil 已提交
480 481 482 483 484 485

/*
 * Reopen a closed connection, with a new peer address.
 */
void ceph_con_open(struct ceph_connection *con, struct ceph_entity_addr *addr)
{
486
	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
S
Sage Weil 已提交
487
	set_bit(OPENING, &con->state);
488 489
	WARN_ON(!test_and_clear_bit(CLOSED, &con->state));

S
Sage Weil 已提交
490
	memcpy(&con->peer_addr, addr, sizeof(*addr));
491
	con->delay = 0;      /* reset backoff memory */
S
Sage Weil 已提交
492 493
	queue_con(con);
}
494
EXPORT_SYMBOL(ceph_con_open);
S
Sage Weil 已提交
495

496 497 498 499 500 501 502 503
/*
 * return true if this connection ever successfully opened
 */
bool ceph_con_opened(struct ceph_connection *con)
{
	return con->connect_seq > 0;
}

S
Sage Weil 已提交
504 505 506 507 508
/*
 * generic get/put
 */
struct ceph_connection *ceph_con_get(struct ceph_connection *con)
{
A
Alex Elder 已提交
509 510 511 512 513
	int nref = __atomic_add_unless(&con->nref, 1, 0);

	dout("con_get %p nref = %d -> %d\n", con, nref, nref + 1);

	return nref ? con : NULL;
S
Sage Weil 已提交
514 515 516 517
}

void ceph_con_put(struct ceph_connection *con)
{
A
Alex Elder 已提交
518 519 520 521
	int nref = atomic_dec_return(&con->nref);

	BUG_ON(nref < 0);
	if (nref == 0) {
522
		BUG_ON(con->sock);
S
Sage Weil 已提交
523 524
		kfree(con);
	}
A
Alex Elder 已提交
525
	dout("con_put %p nref = %d -> %d\n", con, nref + 1, nref);
S
Sage Weil 已提交
526 527 528 529 530
}

/*
 * initialize a new connection.
 */
531 532 533
void ceph_con_init(struct ceph_connection *con, void *private,
	const struct ceph_connection_operations *ops,
	struct ceph_messenger *msgr, __u8 entity_type, __u64 entity_num)
S
Sage Weil 已提交
534 535 536
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
537 538
	con->private = private;
	con->ops = ops;
S
Sage Weil 已提交
539 540
	atomic_set(&con->nref, 1);
	con->msgr = msgr;
541 542 543

	con_sock_state_init(con);

544 545 546
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

547
	mutex_init(&con->mutex);
S
Sage Weil 已提交
548 549 550
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
	INIT_DELAYED_WORK(&con->work, con_work);
551 552

	set_bit(CLOSED, &con->state);
S
Sage Weil 已提交
553
}
554
EXPORT_SYMBOL(ceph_con_init);
S
Sage Weil 已提交
555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572


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

573
static void con_out_kvec_reset(struct ceph_connection *con)
574 575 576 577 578 579
{
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

580
static void con_out_kvec_add(struct ceph_connection *con,
581 582 583 584 585 586 587 588 589 590 591 592
				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 已提交
593 594 595 596 597

/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
598
static void prepare_write_message_footer(struct ceph_connection *con)
S
Sage Weil 已提交
599 600
{
	struct ceph_msg *m = con->out_msg;
601
	int v = con->out_kvec_left;
S
Sage Weil 已提交
602 603 604 605 606 607 608 609

	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;
610
	con->out_msg_done = true;
S
Sage Weil 已提交
611 612 613 614 615 616 617 618
}

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

621
	con_out_kvec_reset(con);
S
Sage Weil 已提交
622
	con->out_kvec_is_msg = true;
623
	con->out_msg_done = false;
S
Sage Weil 已提交
624 625 626 627 628

	/* 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;
629
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
S
Sage Weil 已提交
630
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
631
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
632
			&con->out_temp_ack);
S
Sage Weil 已提交
633 634
	}

635
	BUG_ON(list_empty(&con->out_queue));
636
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
637
	con->out_msg = m;
638
	BUG_ON(m->con != con);
639 640 641 642

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

644 645 646 647 648 649 650 651
	/*
	 * 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;
	}
652 653 654 655
#ifdef CONFIG_BLOCK
	else
		m->bio_iter = NULL;
#endif
S
Sage Weil 已提交
656 657 658 659 660 661 662 663 664

	dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
	     m, con->out_seq, le16_to_cpu(m->hdr.type),
	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
	     le32_to_cpu(m->hdr.data_len),
	     m->nr_pages);
	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

	/* tag + hdr + front + middle */
665 666 667
	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);
668

S
Sage Weil 已提交
669
	if (m->middle)
670
		con_out_kvec_add(con, m->middle->vec.iov_len,
671
			m->middle->vec.iov_base);
S
Sage Weil 已提交
672 673

	/* fill in crc (except data pages), footer */
674 675
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
S
Sage Weil 已提交
676
	con->out_msg->footer.flags = CEPH_MSG_FOOTER_COMPLETE;
677 678 679 680 681 682 683 684

	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 已提交
685 686 687 688 689 690 691 692 693 694
		con->out_msg->footer.middle_crc = 0;
	con->out_msg->footer.data_crc = 0;
	dout("prepare_write_message front_crc %u data_crc %u\n",
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));

	/* is there a data payload? */
	if (le32_to_cpu(m->hdr.data_len) > 0) {
		/* initialize page iterator */
		con->out_msg_pos.page = 0;
695
		if (m->pages)
696
			con->out_msg_pos.page_pos = m->page_alignment;
697 698
		else
			con->out_msg_pos.page_pos = 0;
S
Sage Weil 已提交
699
		con->out_msg_pos.data_pos = 0;
700
		con->out_msg_pos.did_page_crc = false;
S
Sage Weil 已提交
701 702 703
		con->out_more = 1;  /* data + footer will follow */
	} else {
		/* no, queue up footer too and be done */
704
		prepare_write_message_footer(con);
S
Sage Weil 已提交
705 706
	}

707
	set_bit(WRITE_PENDING, &con->flags);
S
Sage Weil 已提交
708 709 710 711 712 713 714 715 716 717 718
}

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

719
	con_out_kvec_reset(con);
720

721
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
722

S
Sage Weil 已提交
723
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
724
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
725 726
				&con->out_temp_ack);

S
Sage Weil 已提交
727
	con->out_more = 1;  /* more will follow.. eventually.. */
728
	set_bit(WRITE_PENDING, &con->flags);
S
Sage Weil 已提交
729 730 731 732 733 734 735 736
}

/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
737 738
	con_out_kvec_reset(con);
	con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
739
	set_bit(WRITE_PENDING, &con->flags);
S
Sage Weil 已提交
740 741 742 743 744 745
}

/*
 * Connection negotiation.
 */

746 747
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
748
{
749
	struct ceph_auth_handshake *auth;
750 751 752 753 754

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;

755
		return NULL;
756 757 758
	}

	/* Can't hold the mutex while getting authorizer */
759

760
	mutex_unlock(&con->mutex);
761

762
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
763

764
	mutex_lock(&con->mutex);
765

766
	if (IS_ERR(auth))
767
		return auth;
768
	if (test_bit(CLOSED, &con->state) || test_bit(OPENING, &con->flags))
769
		return ERR_PTR(-EAGAIN);
770

771 772 773
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;

774

775
	return auth;
776 777
}

S
Sage Weil 已提交
778 779 780
/*
 * We connected to a peer and are saying hello.
 */
781
static void prepare_write_banner(struct ceph_connection *con)
S
Sage Weil 已提交
782
{
783 784
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
785
					&con->msgr->my_enc_addr);
786 787

	con->out_more = 0;
788
	set_bit(WRITE_PENDING, &con->flags);
789 790
}

791
static int prepare_write_connect(struct ceph_connection *con)
792
{
S
Sage Weil 已提交
793 794
	unsigned global_seq = get_global_seq(con->msgr, 0);
	int proto;
795
	int auth_proto;
796
	struct ceph_auth_handshake *auth;
S
Sage Weil 已提交
797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813

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

815
	con->out_connect.features = cpu_to_le64(con->msgr->supported_features);
S
Sage Weil 已提交
816 817 818 819 820 821
	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;

822 823
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
824 825
	if (IS_ERR(auth))
		return PTR_ERR(auth);
826

827
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
828 829 830
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

831
	con_out_kvec_add(con, sizeof (con->out_connect),
832 833
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
834
		con_out_kvec_add(con, auth->authorizer_buf_len,
835
					auth->authorizer_buf);
836

S
Sage Weil 已提交
837
	con->out_more = 0;
838
	set_bit(WRITE_PENDING, &con->flags);
839

840
	return 0;
S
Sage Weil 已提交
841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862
}

/*
 * 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 */
863 864 865 866 867 868 869 870 871 872 873 874

		/* 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 已提交
875 876 877 878 879 880 881 882 883 884 885
		}
	}
	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! */
}

886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910
#ifdef CONFIG_BLOCK
static void init_bio_iter(struct bio *bio, struct bio **iter, int *seg)
{
	if (!bio) {
		*iter = NULL;
		*seg = 0;
		return;
	}
	*iter = bio;
	*seg = bio->bi_idx;
}

static void iter_bio_next(struct bio **bio_iter, int *seg)
{
	if (*bio_iter == NULL)
		return;

	BUG_ON(*seg >= (*bio_iter)->bi_vcnt);

	(*seg)++;
	if (*seg == (*bio_iter)->bi_vcnt)
		init_bio_iter((*bio_iter)->bi_next, bio_iter, seg);
}
#endif

S
Sage Weil 已提交
911 912 913 914 915 916 917 918 919 920 921 922
/*
 * 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
 */
static int write_partial_msg_pages(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->out_msg;
	unsigned data_len = le32_to_cpu(msg->hdr.data_len);
	size_t len;
923
	bool do_datacrc = !con->msgr->nocrc;
S
Sage Weil 已提交
924
	int ret;
925 926 927
	int total_max_write;
	int in_trail = 0;
	size_t trail_len = (msg->trail ? msg->trail->length : 0);
S
Sage Weil 已提交
928 929 930 931 932

	dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
	     con, con->out_msg, con->out_msg_pos.page, con->out_msg->nr_pages,
	     con->out_msg_pos.page_pos);

933 934 935 936 937 938
#ifdef CONFIG_BLOCK
	if (msg->bio && !msg->bio_iter)
		init_bio_iter(msg->bio, &msg->bio_iter, &msg->bio_seg);
#endif

	while (data_len > con->out_msg_pos.data_pos) {
S
Sage Weil 已提交
939
		struct page *page = NULL;
940
		int max_write = PAGE_SIZE;
941
		int bio_offset = 0;
942 943 944

		total_max_write = data_len - trail_len -
			con->out_msg_pos.data_pos;
S
Sage Weil 已提交
945 946 947 948 949 950

		/*
		 * if we are calculating the data crc (the default), we need
		 * to map the page.  if our pages[] has been revoked, use the
		 * zero page.
		 */
951 952 953 954 955 956 957 958 959 960 961

		/* have we reached the trail part of the data? */
		if (con->out_msg_pos.data_pos >= data_len - trail_len) {
			in_trail = 1;

			total_max_write = data_len - con->out_msg_pos.data_pos;

			page = list_first_entry(&msg->trail->head,
						struct page, lru);
			max_write = PAGE_SIZE;
		} else if (msg->pages) {
S
Sage Weil 已提交
962
			page = msg->pages[con->out_msg_pos.page];
963 964 965
		} else if (msg->pagelist) {
			page = list_first_entry(&msg->pagelist->head,
						struct page, lru);
966 967 968 969 970 971
#ifdef CONFIG_BLOCK
		} else if (msg->bio) {
			struct bio_vec *bv;

			bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
			page = bv->bv_page;
972
			bio_offset = bv->bv_offset;
973 974
			max_write = bv->bv_len;
#endif
S
Sage Weil 已提交
975
		} else {
976
			page = zero_page;
S
Sage Weil 已提交
977
		}
978 979 980
		len = min_t(int, max_write - con->out_msg_pos.page_pos,
			    total_max_write);

981
		if (do_datacrc && !con->out_msg_pos.did_page_crc) {
982
			void *base;
983
			u32 crc;
S
Sage Weil 已提交
984
			u32 tmpcrc = le32_to_cpu(con->out_msg->footer.data_crc);
985
			char *kaddr;
S
Sage Weil 已提交
986

987
			kaddr = kmap(page);
S
Sage Weil 已提交
988
			BUG_ON(kaddr == NULL);
989
			base = kaddr + con->out_msg_pos.page_pos + bio_offset;
990 991
			crc = crc32c(tmpcrc, base, len);
			con->out_msg->footer.data_crc = cpu_to_le32(crc);
992
			con->out_msg_pos.did_page_crc = true;
S
Sage Weil 已提交
993
		}
994
		ret = ceph_tcp_sendpage(con->sock, page,
995
				      con->out_msg_pos.page_pos + bio_offset,
996
				      len, 1);
S
Sage Weil 已提交
997

998
		if (do_datacrc)
S
Sage Weil 已提交
999 1000 1001 1002 1003 1004 1005 1006 1007 1008
			kunmap(page);

		if (ret <= 0)
			goto out;

		con->out_msg_pos.data_pos += ret;
		con->out_msg_pos.page_pos += ret;
		if (ret == len) {
			con->out_msg_pos.page_pos = 0;
			con->out_msg_pos.page++;
1009
			con->out_msg_pos.did_page_crc = false;
1010 1011 1012 1013
			if (in_trail)
				list_move_tail(&page->lru,
					       &msg->trail->head);
			else if (msg->pagelist)
1014 1015
				list_move_tail(&page->lru,
					       &msg->pagelist->head);
1016 1017 1018 1019
#ifdef CONFIG_BLOCK
			else if (msg->bio)
				iter_bio_next(&msg->bio_iter, &msg->bio_seg);
#endif
S
Sage Weil 已提交
1020 1021 1022 1023 1024 1025
		}
	}

	dout("write_partial_msg_pages %p msg %p done\n", con, msg);

	/* prepare and queue up footer, too */
1026
	if (!do_datacrc)
S
Sage Weil 已提交
1027
		con->out_msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1028
	con_out_kvec_reset(con);
1029
	prepare_write_message_footer(con);
S
Sage Weil 已提交
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042
	ret = 1;
out:
	return ret;
}

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

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

1045
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, 1);
S
Sage Weil 已提交
1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1058 1059 1060 1061 1062 1063
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096
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;
}

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,
1097
			int end, int size, void *object)
S
Sage Weil 已提交
1098
{
1099 1100
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
S
Sage Weil 已提交
1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113
		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
 */
1114
static int read_partial_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1115
{
1116 1117 1118
	int size;
	int end;
	int ret;
S
Sage Weil 已提交
1119

1120
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
S
Sage Weil 已提交
1121 1122

	/* peer's banner */
1123 1124 1125
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
S
Sage Weil 已提交
1126 1127
	if (ret <= 0)
		goto out;
1128 1129 1130 1131

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
S
Sage Weil 已提交
1132 1133
	if (ret <= 0)
		goto out;
1134 1135 1136 1137

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

1141 1142 1143 1144 1145 1146
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1147 1148 1149
	int size;
	int end;
	int ret;
1150 1151 1152

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

1153 1154 1155
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
S
Sage Weil 已提交
1156 1157
	if (ret <= 0)
		goto out;
1158 1159 1160 1161

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

1165 1166 1167
	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 已提交
1168 1169 1170
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1171

S
Sage Weil 已提交
1172 1173 1174 1175 1176 1177 1178 1179
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1180
		pr_err("connect to %s got bad banner\n",
1181
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206
		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 已提交
1207
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
S
Sage Weil 已提交
1208
	case AF_INET6:
S
Sage Weil 已提交
1209
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
S
Sage Weil 已提交
1210 1211 1212 1213 1214 1215 1216 1217 1218
	}
	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);
1219
		break;
S
Sage Weil 已提交
1220 1221
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1222
		break;
S
Sage Weil 已提交
1223 1224 1225
	}
}

N
Noah Watkins 已提交
1226 1227 1228 1229 1230 1231
/*
 * 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 已提交
1232 1233
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
N
Noah Watkins 已提交
1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320

	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 已提交
1321 1322 1323 1324 1325 1326 1327 1328
/*
 * 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 已提交
1329
	int i, ret = -EINVAL;
S
Sage Weil 已提交
1330 1331 1332 1333 1334 1335 1336
	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 已提交
1337 1338 1339 1340 1341 1342
		char delim = ',';

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

N
Noah Watkins 已提交
1344 1345
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
S
Sage Weil 已提交
1346
			goto bad;
N
Noah Watkins 已提交
1347 1348
		ret = -EINVAL;

S
Sage Weil 已提交
1349 1350
		p = ipend;

S
Sage Weil 已提交
1351 1352 1353 1354 1355 1356 1357 1358
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

S
Sage Weil 已提交
1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374
		/* 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);

1375
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
S
Sage Weil 已提交
1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391

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

1397
static int process_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1398
{
1399
	dout("process_banner on %p\n", con);
S
Sage Weil 已提交
1400 1401 1402 1403

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

1404 1405 1406
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

S
Sage Weil 已提交
1407 1408 1409 1410 1411
	/*
	 * 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 已提交
1412 1413
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
S
Sage Weil 已提交
1414 1415
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
Y
Yehuda Sadeh 已提交
1416
		pr_warning("wrong peer, want %s/%d, got %s/%d\n",
1417
			   ceph_pr_addr(&con->peer_addr.in_addr),
Y
Yehuda Sadeh 已提交
1418
			   (int)le32_to_cpu(con->peer_addr.nonce),
1419
			   ceph_pr_addr(&con->actual_peer_addr.in_addr),
Y
Yehuda Sadeh 已提交
1420
			   (int)le32_to_cpu(con->actual_peer_addr.nonce));
1421
		con->error_msg = "wrong peer at address";
S
Sage Weil 已提交
1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434
		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);
1435
		encode_my_addr(con->msgr);
1436
		dout("process_banner learned my addr is %s\n",
1437
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
S
Sage Weil 已提交
1438 1439
	}

1440 1441 1442 1443 1444
	set_bit(NEGOTIATING, &con->state);
	prepare_read_connect(con);
	return 0;
}

1445 1446 1447 1448 1449 1450
static void fail_protocol(struct ceph_connection *con)
{
	reset_connection(con);
	set_bit(CLOSED, &con->state);  /* in case there's queued work */
}

1451 1452
static int process_connect(struct ceph_connection *con)
{
1453 1454
	u64 sup_feat = con->msgr->supported_features;
	u64 req_feat = con->msgr->required_features;
1455
	u64 server_feat = le64_to_cpu(con->in_reply.features);
1456
	int ret;
1457

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

S
Sage Weil 已提交
1460
	switch (con->in_reply.tag) {
1461 1462 1463 1464
	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),
1465
		       ceph_pr_addr(&con->peer_addr.in_addr),
1466 1467 1468 1469 1470
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
		fail_protocol(con);
		return -1;

S
Sage Weil 已提交
1471 1472 1473 1474
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
1475
		       ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
1476 1477 1478
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
1479
		fail_protocol(con);
S
Sage Weil 已提交
1480 1481
		return -1;

1482 1483 1484 1485 1486 1487 1488 1489 1490
	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;
		}
		con->auth_retry = 1;
1491
		con_out_kvec_reset(con);
1492
		ret = prepare_write_connect(con);
1493 1494
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
1495
		prepare_read_connect(con);
1496
		break;
S
Sage Weil 已提交
1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509

	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",
		     le32_to_cpu(con->in_connect.connect_seq));
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
1510
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1511
		reset_connection(con);
1512
		con_out_kvec_reset(con);
1513 1514 1515
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
1516 1517 1518
		prepare_read_connect(con);

		/* Tell ceph about it. */
1519
		mutex_unlock(&con->mutex);
S
Sage Weil 已提交
1520 1521 1522
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
1523
		mutex_lock(&con->mutex);
1524 1525 1526
		if (test_bit(CLOSED, &con->state) ||
		    test_bit(OPENING, &con->state))
			return -EAGAIN;
S
Sage Weil 已提交
1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537
		break;

	case CEPH_MSGR_TAG_RETRY_SESSION:
		/*
		 * If we sent a smaller connect_seq than the peer has, try
		 * again with a larger value.
		 */
		dout("process_connect got RETRY my seq = %u, peer_seq = %u\n",
		     le32_to_cpu(con->out_connect.connect_seq),
		     le32_to_cpu(con->in_connect.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_connect.connect_seq);
1538
		con_out_kvec_reset(con);
1539 1540 1541
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
1542 1543 1544 1545 1546 1547 1548 1549
		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.
		 */
1550
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
S
Sage Weil 已提交
1551 1552 1553 1554
		     con->peer_global_seq,
		     le32_to_cpu(con->in_connect.global_seq));
		get_global_seq(con->msgr,
			       le32_to_cpu(con->in_connect.global_seq));
1555
		con_out_kvec_reset(con);
1556 1557 1558
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
1559 1560 1561 1562
		prepare_read_connect(con);
		break;

	case CEPH_MSGR_TAG_READY:
1563 1564 1565 1566
		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),
1567
			       ceph_pr_addr(&con->peer_addr.in_addr),
1568 1569 1570 1571 1572
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
			fail_protocol(con);
			return -1;
		}
S
Sage Weil 已提交
1573 1574 1575
		clear_bit(CONNECTING, &con->state);
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
1576
		con->peer_features = server_feat;
S
Sage Weil 已提交
1577 1578 1579 1580 1581 1582
		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 已提交
1583 1584

		if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
1585
			set_bit(LOSSYTX, &con->flags);
S
Sage Weil 已提交
1586

S
Sage Weil 已提交
1587 1588 1589 1590 1591 1592 1593 1594 1595 1596
		prepare_read_tag(con);
		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 已提交
1597 1598 1599
		pr_err("process_connect got WAIT as client\n");
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
S
Sage Weil 已提交
1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614

	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)
{
1615 1616 1617 1618
	int size = sizeof (con->in_temp_ack);
	int end = size;

	return read_partial(con, end, size, &con->in_temp_ack);
S
Sage Weil 已提交
1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638
}


/*
 * 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);
1639
		m->ack_stamp = jiffies;
S
Sage Weil 已提交
1640 1641 1642 1643 1644 1645 1646 1647
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}




1648
static int read_partial_message_section(struct ceph_connection *con,
S
Sage Weil 已提交
1649 1650
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
1651
{
1652
	int ret, left;
1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664

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

1668 1669
	return 1;
}
S
Sage Weil 已提交
1670

A
Alex Elder 已提交
1671 1672
static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
				struct ceph_msg_header *hdr);
1673 1674 1675 1676


static int read_partial_message_pages(struct ceph_connection *con,
				      struct page **pages,
1677
				      unsigned data_len, bool do_datacrc)
1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689
{
	void *p;
	int ret;
	int left;

	left = min((int)(data_len - con->in_msg_pos.data_pos),
		   (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
	/* (page) data */
	BUG_ON(pages == NULL);
	p = kmap(pages[con->in_msg_pos.page]);
	ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
			       left);
1690
	if (ret > 0 && do_datacrc)
1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709
		con->in_data_crc =
			crc32c(con->in_data_crc,
				  p + con->in_msg_pos.page_pos, ret);
	kunmap(pages[con->in_msg_pos.page]);
	if (ret <= 0)
		return ret;
	con->in_msg_pos.data_pos += ret;
	con->in_msg_pos.page_pos += ret;
	if (con->in_msg_pos.page_pos == PAGE_SIZE) {
		con->in_msg_pos.page_pos = 0;
		con->in_msg_pos.page++;
	}

	return ret;
}

#ifdef CONFIG_BLOCK
static int read_partial_message_bio(struct ceph_connection *con,
				    struct bio **bio_iter, int *bio_seg,
1710
				    unsigned data_len, bool do_datacrc)
1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725
{
	struct bio_vec *bv = bio_iovec_idx(*bio_iter, *bio_seg);
	void *p;
	int ret, left;

	if (IS_ERR(bv))
		return PTR_ERR(bv);

	left = min((int)(data_len - con->in_msg_pos.data_pos),
		   (int)(bv->bv_len - con->in_msg_pos.page_pos));

	p = kmap(bv->bv_page) + bv->bv_offset;

	ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
			       left);
1726
	if (ret > 0 && do_datacrc)
1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743
		con->in_data_crc =
			crc32c(con->in_data_crc,
				  p + con->in_msg_pos.page_pos, ret);
	kunmap(bv->bv_page);
	if (ret <= 0)
		return ret;
	con->in_msg_pos.data_pos += ret;
	con->in_msg_pos.page_pos += ret;
	if (con->in_msg_pos.page_pos == bv->bv_len) {
		con->in_msg_pos.page_pos = 0;
		iter_bio_next(bio_iter, bio_seg);
	}

	return ret;
}
#endif

S
Sage Weil 已提交
1744 1745 1746 1747 1748 1749
/*
 * read (part of) a message.
 */
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
1750 1751
	int size;
	int end;
S
Sage Weil 已提交
1752
	int ret;
1753
	unsigned front_len, middle_len, data_len;
1754
	bool do_datacrc = !con->msgr->nocrc;
1755
	u64 seq;
1756
	u32 crc;
S
Sage Weil 已提交
1757 1758 1759 1760

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

	/* header */
1761 1762 1763
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
1764 1765
	if (ret <= 0)
		return ret;
1766 1767 1768 1769 1770 1771 1772 1773 1774

	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 已提交
1775 1776 1777 1778 1779 1780 1781 1782 1783 1784
	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);
	if (middle_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

1785 1786 1787
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
S
Sage Weil 已提交
1788
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1789
			ENTITY_NAME(con->peer_name),
1790
			ceph_pr_addr(&con->peer_addr.in_addr),
1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802
			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 已提交
1803 1804 1805 1806
	/* allocate message? */
	if (!con->in_msg) {
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
		     con->in_hdr.front_len, con->in_hdr.data_len);
A
Alex Elder 已提交
1807
		if (ceph_con_in_msg_alloc(con, &con->in_hdr)) {
S
Sage Weil 已提交
1808
			/* skip this message */
1809
			dout("alloc_msg said skip message\n");
1810
			BUG_ON(con->in_msg);
S
Sage Weil 已提交
1811 1812 1813
			con->in_base_pos = -front_len - middle_len - data_len -
				sizeof(m->footer);
			con->in_tag = CEPH_MSGR_TAG_READY;
1814
			con->in_seq++;
S
Sage Weil 已提交
1815 1816
			return 0;
		}
1817
		if (!con->in_msg) {
1818 1819
			con->error_msg =
				"error allocating memory for incoming message";
1820
			return -ENOMEM;
S
Sage Weil 已提交
1821
		}
1822 1823

		BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
1824 1825
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
1826 1827
		if (m->middle)
			m->middle->vec.iov_len = 0;
1828 1829

		con->in_msg_pos.page = 0;
1830
		if (m->pages)
1831
			con->in_msg_pos.page_pos = m->page_alignment;
1832 1833
		else
			con->in_msg_pos.page_pos = 0;
1834
		con->in_msg_pos.data_pos = 0;
S
Sage Weil 已提交
1835 1836 1837
	}

	/* front */
1838 1839 1840 1841
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
S
Sage Weil 已提交
1842 1843

	/* middle */
1844
	if (m->middle) {
S
Sage Weil 已提交
1845 1846
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
1847
						   &con->in_middle_crc);
S
Sage Weil 已提交
1848 1849 1850
		if (ret <= 0)
			return ret;
	}
1851 1852 1853 1854
#ifdef CONFIG_BLOCK
	if (m->bio && !m->bio_iter)
		init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
#endif
S
Sage Weil 已提交
1855 1856 1857

	/* (page) data */
	while (con->in_msg_pos.data_pos < data_len) {
1858 1859
		if (m->pages) {
			ret = read_partial_message_pages(con, m->pages,
1860
						 data_len, do_datacrc);
1861 1862 1863 1864 1865 1866 1867
			if (ret <= 0)
				return ret;
#ifdef CONFIG_BLOCK
		} else if (m->bio) {

			ret = read_partial_message_bio(con,
						 &m->bio_iter, &m->bio_seg,
1868
						 data_len, do_datacrc);
1869 1870 1871 1872 1873
			if (ret <= 0)
				return ret;
#endif
		} else {
			BUG_ON(1);
S
Sage Weil 已提交
1874 1875 1876 1877
		}
	}

	/* footer */
1878 1879 1880
	size = sizeof (m->footer);
	end += size;
	ret = read_partial(con, end, size, &m->footer);
1881 1882 1883
	if (ret <= 0)
		return ret;

S
Sage Weil 已提交
1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898
	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;
	}
1899
	if (do_datacrc &&
S
Sage Weil 已提交
1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916
	    (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 已提交
1917
	struct ceph_msg *msg;
S
Sage Weil 已提交
1918

1919 1920
	BUG_ON(con->in_msg->con != con);
	con->in_msg->con = NULL;
S
Sage Weil 已提交
1921
	msg = con->in_msg;
S
Sage Weil 已提交
1922
	con->in_msg = NULL;
1923
	ceph_con_put(con);
S
Sage Weil 已提交
1924 1925 1926

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

	con->in_seq++;
1930
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
1931 1932 1933

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
1934
	     ENTITY_NAME(msg->hdr.src),
S
Sage Weil 已提交
1935 1936 1937 1938 1939 1940
	     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);
1941 1942

	mutex_lock(&con->mutex);
S
Sage Weil 已提交
1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962
	prepare_read_tag(con);
}


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

	dout("try_write start %p state %lu nref %d\n", con, con->state,
	     atomic_read(&con->nref));

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

	/* open the socket first? */
	if (con->sock == NULL) {
1963 1964 1965
		clear_bit(NEGOTIATING, &con->state);
		set_bit(CONNECTING, &con->state);

1966
		con_out_kvec_reset(con);
1967
		prepare_write_banner(con);
1968 1969 1970
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
1971
		prepare_read_banner(con);
S
Sage Weil 已提交
1972

1973
		BUG_ON(con->in_msg);
S
Sage Weil 已提交
1974 1975 1976
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
1977 1978
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
S
Sage Weil 已提交
1979 1980 1981 1982 1983 1984 1985 1986 1987 1988
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
	if (con->out_skip) {
		ret = write_partial_skip(con);
		if (ret <= 0)
1989
			goto out;
S
Sage Weil 已提交
1990 1991 1992 1993
	}
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
		if (ret <= 0)
1994
			goto out;
S
Sage Weil 已提交
1995 1996 1997 1998
	}

	/* msg pages? */
	if (con->out_msg) {
1999 2000 2001 2002 2003 2004
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

S
Sage Weil 已提交
2005 2006 2007 2008
		ret = write_partial_msg_pages(con);
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2009
			goto out;
S
Sage Weil 已提交
2010 2011 2012
		if (ret < 0) {
			dout("try_write write_partial_msg_pages err %d\n",
			     ret);
2013
			goto out;
S
Sage Weil 已提交
2014 2015 2016
		}
	}

2017
do_next:
S
Sage Weil 已提交
2018 2019 2020 2021 2022 2023 2024 2025 2026 2027
	if (!test_bit(CONNECTING, &con->state)) {
		/* 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;
		}
2028
		if (test_and_clear_bit(KEEPALIVE_PENDING, &con->flags)) {
S
Sage Weil 已提交
2029 2030 2031 2032 2033 2034
			prepare_write_keepalive(con);
			goto more;
		}
	}

	/* Nothing to do! */
2035
	clear_bit(WRITE_PENDING, &con->flags);
S
Sage Weil 已提交
2036 2037 2038
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2039
	dout("try_write done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058
	return ret;
}



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

	if (!con->sock)
		return 0;

	if (test_bit(STANDBY, &con->state))
		return 0;

	dout("try_read start on %p\n", con);
2059

S
Sage Weil 已提交
2060 2061 2062
more:
	dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
	     con->in_base_pos);
2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073

	/*
	 * process_connect and process_message drop and re-take
	 * con->mutex.  make sure we handle a racing close or reopen.
	 */
	if (test_bit(CLOSED, &con->state) ||
	    test_bit(OPENING, &con->state)) {
		ret = -EAGAIN;
		goto out;
	}

S
Sage Weil 已提交
2074
	if (test_bit(CONNECTING, &con->state)) {
2075 2076 2077 2078 2079
		if (!test_bit(NEGOTIATING, &con->state)) {
			dout("try_read connecting\n");
			ret = read_partial_banner(con);
			if (ret <= 0)
				goto out;
2080 2081 2082
			ret = process_banner(con);
			if (ret < 0)
				goto out;
2083
		}
S
Sage Weil 已提交
2084 2085 2086
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2087 2088 2089
		ret = process_connect(con);
		if (ret < 0)
			goto out;
S
Sage Weil 已提交
2090 2091 2092 2093 2094 2095 2096 2097 2098
		goto more;
	}

	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
A
Alex Elder 已提交
2099 2100 2101
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

S
Sage Weil 已提交
2102 2103 2104
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2105
			goto out;
S
Sage Weil 已提交
2106 2107 2108 2109 2110 2111 2112 2113 2114 2115
		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)
2116
			goto out;
S
Sage Weil 已提交
2117 2118 2119 2120 2121 2122 2123 2124 2125 2126
		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:
			set_bit(CLOSED, &con->state);   /* fixme */
2127
			goto out;
S
Sage Weil 已提交
2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138
		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;
2139
				break;
S
Sage Weil 已提交
2140 2141
			case -EIO:
				con->error_msg = "io error";
2142
				break;
S
Sage Weil 已提交
2143
			}
2144
			goto out;
S
Sage Weil 已提交
2145 2146 2147 2148 2149 2150 2151 2152 2153
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
		goto more;
	}
	if (con->in_tag == CEPH_MSGR_TAG_ACK) {
		ret = read_partial_ack(con);
		if (ret <= 0)
2154
			goto out;
S
Sage Weil 已提交
2155 2156 2157 2158 2159
		process_ack(con);
		goto more;
	}

out:
2160
	dout("try_read done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181
	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;
}


/*
 * Atomically queue work on a connection.  Bump @con reference to
 * avoid races with connection teardown.
 */
static void queue_con(struct ceph_connection *con)
{
	if (!con->ops->get(con)) {
		dout("queue_con %p ref count 0\n", con);
		return;
	}

2182
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
S
Sage Weil 已提交
2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196
		dout("queue_con %p - already queued\n", con);
		con->ops->put(con);
	} else {
		dout("queue_con %p\n", con);
	}
}

/*
 * 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);
2197
	int ret;
S
Sage Weil 已提交
2198

S
Sage Weil 已提交
2199
	mutex_lock(&con->mutex);
2200
restart:
2201
	if (test_and_clear_bit(BACKOFF, &con->flags)) {
S
Sage Weil 已提交
2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213
		dout("con_work %p backing off\n", con);
		if (queue_delayed_work(ceph_msgr_wq, &con->work,
				       round_jiffies_relative(con->delay))) {
			dout("con_work %p backoff %lu\n", con, con->delay);
			mutex_unlock(&con->mutex);
			return;
		} else {
			con->ops->put(con);
			dout("con_work %p FAILED to back off %lu\n", con,
			     con->delay);
		}
	}
S
Sage Weil 已提交
2214

S
Sage Weil 已提交
2215 2216 2217 2218
	if (test_bit(STANDBY, &con->state)) {
		dout("con_work %p STANDBY\n", con);
		goto done;
	}
S
Sage Weil 已提交
2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229
	if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
		dout("con_work CLOSED\n");
		con_close_socket(con);
		goto done;
	}
	if (test_and_clear_bit(OPENING, &con->state)) {
		/* reopen w/ new peer */
		dout("con_work OPENING\n");
		con_close_socket(con);
	}

2230
	if (test_and_clear_bit(SOCK_CLOSED, &con->flags))
2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243
		goto fault;

	ret = try_read(con);
	if (ret == -EAGAIN)
		goto restart;
	if (ret < 0)
		goto fault;

	ret = try_write(con);
	if (ret == -EAGAIN)
		goto restart;
	if (ret < 0)
		goto fault;
S
Sage Weil 已提交
2244 2245

done:
S
Sage Weil 已提交
2246 2247
	mutex_unlock(&con->mutex);
done_unlocked:
S
Sage Weil 已提交
2248
	con->ops->put(con);
2249 2250 2251 2252 2253 2254
	return;

fault:
	mutex_unlock(&con->mutex);
	ceph_fault(con);     /* error/fault path */
	goto done_unlocked;
S
Sage Weil 已提交
2255 2256 2257 2258 2259 2260 2261 2262 2263 2264
}


/*
 * Generic error/fault handler.  A retry mechanism is used with
 * exponential backoff
 */
static void ceph_fault(struct ceph_connection *con)
{
	pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2265
	       ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
S
Sage Weil 已提交
2266
	dout("fault %p state %lu to peer %s\n",
2267
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2268

2269
	if (test_bit(LOSSYTX, &con->flags)) {
S
Sage Weil 已提交
2270 2271 2272 2273
		dout("fault on LOSSYTX channel\n");
		goto out;
	}

2274
	mutex_lock(&con->mutex);
2275 2276
	if (test_bit(CLOSED, &con->state))
		goto out_unlock;
2277

S
Sage Weil 已提交
2278
	con_close_socket(con);
S
Sage Weil 已提交
2279 2280

	if (con->in_msg) {
2281 2282
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
S
Sage Weil 已提交
2283 2284
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
2285
		ceph_con_put(con);
S
Sage Weil 已提交
2286
	}
S
Sage Weil 已提交
2287

2288 2289
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2290

S
Sage Weil 已提交
2291 2292 2293
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2294
	    !test_bit(KEEPALIVE_PENDING, &con->flags)) {
S
Sage Weil 已提交
2295
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2296
		clear_bit(WRITE_PENDING, &con->flags);
S
Sage Weil 已提交
2297
		set_bit(STANDBY, &con->state);
2298 2299 2300 2301 2302 2303 2304 2305
	} else {
		/* retry after a delay. */
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
		con->ops->get(con);
		if (queue_delayed_work(ceph_msgr_wq, &con->work,
S
Sage Weil 已提交
2306 2307 2308
				       round_jiffies_relative(con->delay))) {
			dout("fault queued %p delay %lu\n", con, con->delay);
		} else {
2309
			con->ops->put(con);
S
Sage Weil 已提交
2310 2311 2312 2313 2314 2315 2316 2317 2318 2319
			dout("fault failed to queue %p delay %lu, backoff\n",
			     con, con->delay);
			/*
			 * In many cases we see a socket state change
			 * while con_work is running and end up
			 * queuing (non-delayed) work, such that we
			 * can't backoff with a delay.  Set a flag so
			 * that when con_work restarts we schedule the
			 * delay then.
			 */
2320
			set_bit(BACKOFF, &con->flags);
S
Sage Weil 已提交
2321
		}
S
Sage Weil 已提交
2322 2323
	}

2324 2325
out_unlock:
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2326
out:
2327 2328 2329
	/*
	 * in case we faulted due to authentication, invalidate our
	 * current tickets so that we can get new ones.
S
Sage Weil 已提交
2330
	 */
2331 2332 2333 2334 2335
	if (con->auth_retry && con->ops->invalidate_authorizer) {
		dout("calling invalidate_authorizer()\n");
		con->ops->invalidate_authorizer(con);
	}

S
Sage Weil 已提交
2336 2337 2338 2339 2340 2341 2342
	if (con->ops->fault)
		con->ops->fault(con);
}



/*
2343
 * initialize a new messenger instance
S
Sage Weil 已提交
2344
 */
2345 2346 2347 2348 2349
void ceph_messenger_init(struct ceph_messenger *msgr,
			struct ceph_entity_addr *myaddr,
			u32 supported_features,
			u32 required_features,
			bool nocrc)
S
Sage Weil 已提交
2350
{
2351 2352 2353
	msgr->supported_features = supported_features;
	msgr->required_features = required_features;

S
Sage Weil 已提交
2354 2355 2356 2357 2358 2359
	spin_lock_init(&msgr->global_seq_lock);

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

	/* select a random nonce */
2360
	msgr->inst.addr.type = 0;
S
Sage Weil 已提交
2361
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2362
	encode_my_addr(msgr);
2363
	msgr->nocrc = nocrc;
S
Sage Weil 已提交
2364

2365
	dout("%s %p\n", __func__, msgr);
S
Sage Weil 已提交
2366
}
2367
EXPORT_SYMBOL(ceph_messenger_init);
S
Sage Weil 已提交
2368

S
Sage Weil 已提交
2369 2370 2371 2372 2373 2374 2375
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
	if (test_and_clear_bit(STANDBY, &con->state)) {
		mutex_lock(&con->mutex);
		dout("clear_standby %p and ++connect_seq\n", con);
		con->connect_seq++;
2376 2377
		WARN_ON(test_bit(WRITE_PENDING, &con->flags));
		WARN_ON(test_bit(KEEPALIVE_PENDING, &con->flags));
S
Sage Weil 已提交
2378 2379 2380 2381
		mutex_unlock(&con->mutex);
	}
}

S
Sage Weil 已提交
2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	if (test_bit(CLOSED, &con->state)) {
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		return;
	}

	/* set src+dst */
2394
	msg->hdr.src = con->msgr->inst.name;
S
Sage Weil 已提交
2395

2396 2397
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));

2398 2399
	msg->needs_out_seq = true;

S
Sage Weil 已提交
2400
	/* queue */
2401
	mutex_lock(&con->mutex);
2402

2403
	BUG_ON(msg->con != NULL);
2404 2405 2406
	msg->con = ceph_con_get(con);
	BUG_ON(msg->con == NULL);

S
Sage Weil 已提交
2407 2408 2409 2410 2411 2412 2413 2414
	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));
2415
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2416 2417 2418

	/* if there wasn't anything waiting to send before, queue
	 * new work */
S
Sage Weil 已提交
2419
	clear_standby(con);
2420
	if (test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
S
Sage Weil 已提交
2421 2422
		queue_con(con);
}
2423
EXPORT_SYMBOL(ceph_con_send);
S
Sage Weil 已提交
2424 2425 2426 2427

/*
 * Revoke a message that was previously queued for send
 */
2428
void ceph_msg_revoke(struct ceph_msg *msg)
S
Sage Weil 已提交
2429
{
2430 2431 2432 2433 2434
	struct ceph_connection *con = msg->con;

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

2435
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
2436
	if (!list_empty(&msg->list_head)) {
2437
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
S
Sage Weil 已提交
2438
		list_del_init(&msg->list_head);
2439
		BUG_ON(msg->con == NULL);
2440
		ceph_con_put(msg->con);
2441
		msg->con = NULL;
2442
		msg->hdr.seq = 0;
2443

S
Sage Weil 已提交
2444
		ceph_msg_put(msg);
S
Sage Weil 已提交
2445 2446
	}
	if (con->out_msg == msg) {
2447
		dout("%s %p msg %p - was sending\n", __func__, con, msg);
S
Sage Weil 已提交
2448
		con->out_msg = NULL;
S
Sage Weil 已提交
2449 2450 2451 2452
		if (con->out_kvec_is_msg) {
			con->out_skip = con->out_kvec_bytes;
			con->out_kvec_is_msg = false;
		}
S
Sage Weil 已提交
2453
		msg->hdr.seq = 0;
2454 2455

		ceph_msg_put(msg);
S
Sage Weil 已提交
2456
	}
2457
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2458 2459
}

2460
/*
2461
 * Revoke a message that we may be reading data into
2462
 */
2463
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2464
{
2465 2466 2467 2468 2469 2470 2471 2472 2473 2474
	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;
2475
	mutex_lock(&con->mutex);
2476
	if (con->in_msg == msg) {
2477 2478
		unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
		unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
2479 2480 2481
		unsigned data_len = le32_to_cpu(con->in_hdr.data_len);

		/* skip rest of message */
2482 2483
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
2484
				sizeof(struct ceph_msg_header) -
2485 2486 2487
				front_len -
				middle_len -
				data_len -
2488 2489 2490 2491
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
2492
		con->in_seq++;
2493
	} else {
2494 2495
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
2496 2497 2498 2499
	}
	mutex_unlock(&con->mutex);
}

S
Sage Weil 已提交
2500 2501 2502 2503 2504
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
S
Sage Weil 已提交
2505 2506
	dout("con_keepalive %p\n", con);
	clear_standby(con);
2507 2508
	if (test_and_set_bit(KEEPALIVE_PENDING, &con->flags) == 0 &&
	    test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
S
Sage Weil 已提交
2509 2510
		queue_con(con);
}
2511
EXPORT_SYMBOL(ceph_con_keepalive);
S
Sage Weil 已提交
2512 2513 2514 2515 2516 2517


/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
2518 2519
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
S
Sage Weil 已提交
2520 2521 2522
{
	struct ceph_msg *m;

2523
	m = kmalloc(sizeof(*m), flags);
S
Sage Weil 已提交
2524 2525
	if (m == NULL)
		goto out;
S
Sage Weil 已提交
2526
	kref_init(&m->kref);
2527 2528

	m->con = NULL;
S
Sage Weil 已提交
2529 2530
	INIT_LIST_HEAD(&m->list_head);

2531
	m->hdr.tid = 0;
S
Sage Weil 已提交
2532
	m->hdr.type = cpu_to_le16(type);
2533 2534
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
	m->hdr.version = 0;
S
Sage Weil 已提交
2535 2536
	m->hdr.front_len = cpu_to_le32(front_len);
	m->hdr.middle_len = 0;
S
Sage Weil 已提交
2537 2538
	m->hdr.data_len = 0;
	m->hdr.data_off = 0;
2539
	m->hdr.reserved = 0;
S
Sage Weil 已提交
2540 2541 2542
	m->footer.front_crc = 0;
	m->footer.middle_crc = 0;
	m->footer.data_crc = 0;
2543
	m->footer.flags = 0;
S
Sage Weil 已提交
2544 2545 2546
	m->front_max = front_len;
	m->front_is_vmalloc = false;
	m->more_to_follow = false;
2547
	m->ack_stamp = 0;
S
Sage Weil 已提交
2548 2549
	m->pool = NULL;

2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562
	/* middle */
	m->middle = NULL;

	/* data */
	m->nr_pages = 0;
	m->page_alignment = 0;
	m->pages = NULL;
	m->pagelist = NULL;
	m->bio = NULL;
	m->bio_iter = NULL;
	m->bio_seg = 0;
	m->trail = NULL;

S
Sage Weil 已提交
2563 2564 2565
	/* front */
	if (front_len) {
		if (front_len > PAGE_CACHE_SIZE) {
2566
			m->front.iov_base = __vmalloc(front_len, flags,
S
Sage Weil 已提交
2567 2568 2569
						      PAGE_KERNEL);
			m->front_is_vmalloc = true;
		} else {
2570
			m->front.iov_base = kmalloc(front_len, flags);
S
Sage Weil 已提交
2571 2572
		}
		if (m->front.iov_base == NULL) {
2573
			dout("ceph_msg_new can't allocate %d bytes\n",
S
Sage Weil 已提交
2574 2575 2576 2577 2578 2579 2580 2581
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
	m->front.iov_len = front_len;

S
Sage Weil 已提交
2582
	dout("ceph_msg_new %p front %d\n", m, front_len);
S
Sage Weil 已提交
2583 2584 2585 2586 2587
	return m;

out2:
	ceph_msg_put(m);
out:
2588 2589 2590
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
2591
		WARN_ON(1);
2592 2593 2594 2595
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
2596
	return NULL;
S
Sage Weil 已提交
2597
}
2598
EXPORT_SYMBOL(ceph_msg_new);
S
Sage Weil 已提交
2599 2600 2601 2602 2603 2604 2605 2606

/*
 * 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.
 */
2607
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
S
Sage Weil 已提交
2608 2609 2610 2611 2612 2613 2614 2615 2616
{
	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 已提交
2617
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
S
Sage Weil 已提交
2618 2619 2620 2621 2622
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

2623
/*
A
Alex Elder 已提交
2624 2625 2626 2627 2628 2629 2630 2631
 * 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.
 *
 * Returns true if the message should be skipped, false otherwise.
 * If true is returned (skip message), con->in_msg will be NULL.
 * If false is returned, con->in_msg will contain a pointer to the
 * newly-allocated message, or NULL in case of memory exhaustion.
2632
 */
A
Alex Elder 已提交
2633 2634
static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
				struct ceph_msg_header *hdr)
2635 2636 2637 2638 2639 2640
{
	int type = le16_to_cpu(hdr->type);
	int front_len = le32_to_cpu(hdr->front_len);
	int middle_len = le32_to_cpu(hdr->middle_len);
	int ret;

A
Alex Elder 已提交
2641 2642
	BUG_ON(con->in_msg != NULL);

2643
	if (con->ops->alloc_msg) {
A
Alex Elder 已提交
2644 2645
		int skip = 0;

2646
		mutex_unlock(&con->mutex);
A
Alex Elder 已提交
2647
		con->in_msg = con->ops->alloc_msg(con, hdr, &skip);
2648
		mutex_lock(&con->mutex);
2649 2650 2651 2652
		if (con->in_msg) {
			con->in_msg->con = ceph_con_get(con);
			BUG_ON(con->in_msg->con == NULL);
		}
A
Alex Elder 已提交
2653 2654 2655 2656 2657
		if (skip)
			con->in_msg = NULL;

		if (!con->in_msg)
			return skip != 0;
2658
	}
A
Alex Elder 已提交
2659 2660 2661
	if (!con->in_msg) {
		con->in_msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
		if (!con->in_msg) {
2662 2663
			pr_err("unable to allocate msg type %d len %d\n",
			       type, front_len);
A
Alex Elder 已提交
2664
			return false;
2665
		}
2666 2667
		con->in_msg->con = ceph_con_get(con);
		BUG_ON(con->in_msg->con == NULL);
A
Alex Elder 已提交
2668
		con->in_msg->page_alignment = le16_to_cpu(hdr->data_off);
2669
	}
A
Alex Elder 已提交
2670
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2671

A
Alex Elder 已提交
2672 2673
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
2674
		if (ret < 0) {
A
Alex Elder 已提交
2675 2676
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
2677 2678
		}
	}
2679

A
Alex Elder 已提交
2680
	return false;
2681 2682
}

S
Sage Weil 已提交
2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699

/*
 * 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 已提交
2700 2701 2702
void ceph_msg_last_put(struct kref *kref)
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
S
Sage Weil 已提交
2703

S
Sage Weil 已提交
2704 2705 2706 2707 2708 2709 2710
	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 已提交
2711
	}
S
Sage Weil 已提交
2712 2713 2714
	m->nr_pages = 0;
	m->pages = NULL;

2715 2716 2717 2718 2719 2720
	if (m->pagelist) {
		ceph_pagelist_release(m->pagelist);
		kfree(m->pagelist);
		m->pagelist = NULL;
	}

2721 2722
	m->trail = NULL;

S
Sage Weil 已提交
2723 2724 2725 2726
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
		ceph_msg_kfree(m);
S
Sage Weil 已提交
2727
}
2728
EXPORT_SYMBOL(ceph_msg_last_put);
2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743 2744 2745 2746 2747 2748

void ceph_msg_dump(struct ceph_msg *msg)
{
	pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
		 msg->front_max, msg->nr_pages);
	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);
}
2749
EXPORT_SYMBOL(ceph_msg_dump);