messenger.c 66.6 KB
Newer Older
1
#include <linux/ceph/ceph_debug.h>
Sage Weil's avatar
Sage Weil committed
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>
Sage Weil's avatar
Sage Weil committed
10 11
#include <linux/socket.h>
#include <linux/string.h>
12 13
#include <linux/bio.h>
#include <linux/blkdev.h>
Noah Watkins's avatar
Noah Watkins committed
14
#include <linux/dns_resolver.h>
Sage Weil's avatar
Sage Weil committed
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>
Sage Weil's avatar
Sage Weil committed
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 */

Sage Weil's avatar
Sage Weil committed
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

Alex Elder's avatar
Alex Elder committed
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
Sage Weil's avatar
Sage Weil committed
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.
Sage Weil's avatar
Sage Weil committed
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);
Sage Weil's avatar
Sage Weil committed
70

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

73
const char *ceph_pr_addr(const struct sockaddr_storage *ss)
Sage Weil's avatar
Sage Weil committed
74 75 76
{
	int i;
	char *s;
77 78
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
Sage Weil's avatar
Sage Weil committed
79

80
	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
Sage Weil's avatar
Sage Weil committed
81 82 83 84
	s = addr_str[i];

	switch (ss->ss_family) {
	case AF_INET:
85 86
		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
			 ntohs(in4->sin_port));
Sage Weil's avatar
Sage Weil committed
87 88 89
		break;

	case AF_INET6:
90 91
		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
			 ntohs(in6->sin6_port));
Sage Weil's avatar
Sage Weil committed
92 93 94
		break;

	default:
Alex Elder's avatar
Alex Elder committed
95 96
		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
			 ss->ss_family);
Sage Weil's avatar
Sage Weil committed
97 98 99 100
	}

	return s;
}
101
EXPORT_SYMBOL(ceph_pr_addr);
Sage Weil's avatar
Sage Weil committed
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);
}

Sage Weil's avatar
Sage Weil committed
109 110 111
/*
 * work queue for all reading and writing to/from the socket.
 */
112
static struct workqueue_struct *ceph_msgr_wq;
Sage Weil's avatar
Sage Weil committed
113

114 115
void _ceph_msgr_exit(void)
{
Alex Elder's avatar
Alex Elder committed
116
	if (ceph_msgr_wq) {
117
		destroy_workqueue(ceph_msgr_wq);
Alex Elder's avatar
Alex Elder committed
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)
Sage Weil's avatar
Sage Weil committed
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;
Sage Weil's avatar
Sage Weil committed
141
}
142
EXPORT_SYMBOL(ceph_msgr_init);
Sage Weil's avatar
Sage Weil committed
143 144 145

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

148
	_ceph_msgr_exit();
Sage Weil's avatar
Sage Weil committed
149
}
150
EXPORT_SYMBOL(ceph_msgr_exit);
Sage Weil's avatar
Sage Weil committed
151

Yehuda Sadeh's avatar
Yehuda Sadeh committed
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

Sage Weil's avatar
Sage Weil committed
208 209 210 211 212
/*
 * socket callback functions
 */

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

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

/* socket has buffer space for writing */
Alex Elder's avatar
Alex Elder committed
225
static void ceph_sock_write_space(struct sock *sk)
Sage Weil's avatar
Sage Weil committed
226
{
Alex Elder's avatar
Alex Elder committed
227
	struct ceph_connection *con = sk->sk_user_data;
Sage Weil's avatar
Sage Weil committed
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
Alex Elder's avatar
Alex Elder committed
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)) {
Alex Elder's avatar
Alex Elder committed
238
			dout("%s %p queueing write work\n", __func__, con);
239 240 241
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
Sage Weil's avatar
Sage Weil committed
242
	} else {
Alex Elder's avatar
Alex Elder committed
243
		dout("%s %p nothing to write\n", __func__, con);
Sage Weil's avatar
Sage Weil committed
244 245 246 247
	}
}

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

Alex Elder's avatar
Alex Elder committed
252
	dout("%s %p state = %lu sk_state = %u\n", __func__,
Sage Weil's avatar
Sage Weil committed
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:
Alex Elder's avatar
Alex Elder committed
260
		dout("%s TCP_CLOSE\n", __func__);
Sage Weil's avatar
Sage Weil committed
261
	case TCP_CLOSE_WAIT:
Alex Elder's avatar
Alex Elder committed
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) {
Sage Weil's avatar
Sage Weil committed
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:
Alex Elder's avatar
Alex Elder committed
273
		dout("%s TCP_ESTABLISHED\n", __func__);
274
		con_sock_state_connected(con);
Sage Weil's avatar
Sage Weil committed
275 276
		queue_con(con);
		break;
Alex Elder's avatar
Alex Elder committed
277 278
	default:	/* Everything else is uninteresting */
		break;
Sage Weil's avatar
Sage Weil committed
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;
289
	sk->sk_user_data = con;
Alex Elder's avatar
Alex Elder committed
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;
Sage Weil's avatar
Sage Weil committed
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)
Sage Weil's avatar
Sage Weil committed
304
{
Sage Weil's avatar
Sage Weil committed
305
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
Sage Weil's avatar
Sage Weil committed
306 307 308 309
	struct socket *sock;
	int ret;

	BUG_ON(con->sock);
Sage Weil's avatar
Sage Weil committed
310 311
	ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
			       IPPROTO_TCP, &sock);
Sage Weil's avatar
Sage Weil committed
312
	if (ret)
313
		return ret;
Sage Weil's avatar
Sage Weil committed
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

Sage Weil's avatar
Sage Weil committed
320 321
	set_sock_callbacks(sock, con);

322
	dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
323

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

337
		return ret;
338 339
	}
	con->sock = sock;
340
	return 0;
Sage Weil's avatar
Sage Weil committed
341 342 343 344 345 346
}

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 };
347
	int r;
Sage Weil's avatar
Sage Weil committed
348

349 350 351 352
	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
Sage Weil's avatar
Sage Weil committed
353 354 355 356 357 358 359 360 361 362
}

/*
 * 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 };
363
	int r;
Sage Weil's avatar
Sage Weil committed
364 365 366 367 368 369

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

370 371 372 373
	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
	if (r == -EAGAIN)
		r = 0;
	return r;
Sage Weil's avatar
Sage Weil committed
374 375
}

376 377 378 379 380 381 382 383 384 385 386 387 388
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;
}

Sage Weil's avatar
Sage Weil committed
389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404

/*
 * 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);
405
	con_sock_state_closed(con);
Sage Weil's avatar
Sage Weil committed
406 407 408 409 410 411 412 413 414 415
	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);
416
	BUG_ON(msg->con == NULL);
Sage Weil's avatar
Sage Weil committed
417
	msg->con->ops->put(msg->con);
418 419
	msg->con = NULL;

Sage Weil's avatar
Sage Weil committed
420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437
	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);

438
	if (con->in_msg) {
439 440
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
441 442
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
Sage Weil's avatar
Sage Weil committed
443
		con->ops->put(con);
444 445
	}

Sage Weil's avatar
Sage Weil committed
446 447
	con->connect_seq = 0;
	con->out_seq = 0;
448 449 450 451
	if (con->out_msg) {
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
Sage Weil's avatar
Sage Weil committed
452
	con->in_seq = 0;
453
	con->in_seq_acked = 0;
Sage Weil's avatar
Sage Weil committed
454 455 456 457 458 459 460
}

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

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

471
	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
472
	reset_connection(con);
Sage Weil's avatar
Sage Weil committed
473
	con->peer_global_seq = 0;
474
	cancel_delayed_work(&con->work);
475
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
476 477
	queue_con(con);
}
478
EXPORT_SYMBOL(ceph_con_close);
Sage Weil's avatar
Sage Weil committed
479 480 481 482 483 484

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

Sage Weil's avatar
Sage Weil committed
489
	memcpy(&con->peer_addr, addr, sizeof(*addr));
490
	con->delay = 0;      /* reset backoff memory */
Sage Weil's avatar
Sage Weil committed
491 492
	queue_con(con);
}
493
EXPORT_SYMBOL(ceph_con_open);
Sage Weil's avatar
Sage Weil committed
494

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

Sage Weil's avatar
Sage Weil committed
503 504 505
/*
 * initialize a new connection.
 */
506 507 508
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)
Sage Weil's avatar
Sage Weil committed
509 510 511
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
512 513
	con->private = private;
	con->ops = ops;
Sage Weil's avatar
Sage Weil committed
514
	con->msgr = msgr;
515 516 517

	con_sock_state_init(con);

518 519 520
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

521
	mutex_init(&con->mutex);
Sage Weil's avatar
Sage Weil committed
522 523 524
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
	INIT_DELAYED_WORK(&con->work, con_work);
525 526

	set_bit(CLOSED, &con->state);
Sage Weil's avatar
Sage Weil committed
527
}
528
EXPORT_SYMBOL(ceph_con_init);
Sage Weil's avatar
Sage Weil committed
529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546


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

547
static void con_out_kvec_reset(struct ceph_connection *con)
548 549 550 551 552 553
{
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

554
static void con_out_kvec_add(struct ceph_connection *con,
555 556 557 558 559 560 561 562 563 564 565 566
				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;
}
Sage Weil's avatar
Sage Weil committed
567

568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592
#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

593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610
static void prepare_write_message_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->out_msg;

	BUG_ON(!msg);
	BUG_ON(!msg->hdr.data_len);

	/* initialize page iterator */
	con->out_msg_pos.page = 0;
	if (msg->pages)
		con->out_msg_pos.page_pos = msg->page_alignment;
	else
		con->out_msg_pos.page_pos = 0;
	con->out_msg_pos.data_pos = 0;
	con->out_msg_pos.did_page_crc = false;
	con->out_more = 1;  /* data + footer will follow */
}

Sage Weil's avatar
Sage Weil committed
611 612 613 614
/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
615
static void prepare_write_message_footer(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
616 617
{
	struct ceph_msg *m = con->out_msg;
618
	int v = con->out_kvec_left;
Sage Weil's avatar
Sage Weil committed
619

620 621
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

Sage Weil's avatar
Sage Weil committed
622 623 624 625 626 627 628
	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;
629
	con->out_msg_done = true;
Sage Weil's avatar
Sage Weil committed
630 631 632 633 634 635 636 637
}

/*
 * Prepare headers for the next outgoing message.
 */
static void prepare_write_message(struct ceph_connection *con)
{
	struct ceph_msg *m;
638
	u32 crc;
Sage Weil's avatar
Sage Weil committed
639

640
	con_out_kvec_reset(con);
Sage Weil's avatar
Sage Weil committed
641
	con->out_kvec_is_msg = true;
642
	con->out_msg_done = false;
Sage Weil's avatar
Sage Weil committed
643 644 645 646 647

	/* 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;
648
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
Sage Weil's avatar
Sage Weil committed
649
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
650
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
651
			&con->out_temp_ack);
Sage Weil's avatar
Sage Weil committed
652 653
	}

654
	BUG_ON(list_empty(&con->out_queue));
655
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
656
	con->out_msg = m;
657
	BUG_ON(m->con != con);
658 659 660 661

	/* put message on sent list */
	ceph_msg_get(m);
	list_move_tail(&m->list_head, &con->out_sent);
Sage Weil's avatar
Sage Weil committed
662

663 664 665 666 667 668 669 670
	/*
	 * 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;
	}
671 672 673 674
#ifdef CONFIG_BLOCK
	else
		m->bio_iter = NULL;
#endif
Sage Weil's avatar
Sage Weil committed
675 676 677 678 679 680 681 682 683

	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 */
684 685 686
	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);
687

Sage Weil's avatar
Sage Weil committed
688
	if (m->middle)
689
		con_out_kvec_add(con, m->middle->vec.iov_len,
690
			m->middle->vec.iov_base);
Sage Weil's avatar
Sage Weil committed
691 692

	/* fill in crc (except data pages), footer */
693 694
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
695
	con->out_msg->footer.flags = 0;
696 697 698 699 700 701 702 703

	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
Sage Weil's avatar
Sage Weil committed
704
		con->out_msg->footer.middle_crc = 0;
705
	dout("%s front_crc %u middle_crc %u\n", __func__,
Sage Weil's avatar
Sage Weil committed
706 707 708 709
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));

	/* is there a data payload? */
710 711 712 713
	con->out_msg->footer.data_crc = 0;
	if (m->hdr.data_len)
		prepare_write_message_data(con);
	else
Sage Weil's avatar
Sage Weil committed
714
		/* no, queue up footer too and be done */
715
		prepare_write_message_footer(con);
Sage Weil's avatar
Sage Weil committed
716

717
	set_bit(WRITE_PENDING, &con->flags);
Sage Weil's avatar
Sage Weil committed
718 719 720 721 722 723 724 725 726 727 728
}

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

729
	con_out_kvec_reset(con);
730

731
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
732

Sage Weil's avatar
Sage Weil committed
733
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
734
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
735 736
				&con->out_temp_ack);

Sage Weil's avatar
Sage Weil committed
737
	con->out_more = 1;  /* more will follow.. eventually.. */
738
	set_bit(WRITE_PENDING, &con->flags);
Sage Weil's avatar
Sage Weil committed
739 740 741 742 743 744 745 746
}

/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
747 748
	con_out_kvec_reset(con);
	con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
749
	set_bit(WRITE_PENDING, &con->flags);
Sage Weil's avatar
Sage Weil committed
750 751 752 753 754 755
}

/*
 * Connection negotiation.
 */

756 757
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
758
{
759
	struct ceph_auth_handshake *auth;
760 761 762 763 764

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

765
		return NULL;
766 767 768
	}

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

770
	mutex_unlock(&con->mutex);
771

772
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
773

774
	mutex_lock(&con->mutex);
775

776
	if (IS_ERR(auth))
777
		return auth;
778
	if (test_bit(CLOSED, &con->state) || test_bit(OPENING, &con->flags))
779
		return ERR_PTR(-EAGAIN);
780

781 782 783
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;

784

785
	return auth;
786 787
}

Sage Weil's avatar
Sage Weil committed
788 789 790
/*
 * We connected to a peer and are saying hello.
 */
791
static void prepare_write_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
792
{
793 794
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
795
					&con->msgr->my_enc_addr);
796 797

	con->out_more = 0;
798
	set_bit(WRITE_PENDING, &con->flags);
799 800
}

801
static int prepare_write_connect(struct ceph_connection *con)
802
{
803
	unsigned int global_seq = get_global_seq(con->msgr, 0);
Sage Weil's avatar
Sage Weil committed
804
	int proto;
805
	int auth_proto;
806
	struct ceph_auth_handshake *auth;
Sage Weil's avatar
Sage Weil committed
807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823

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

825
	con->out_connect.features = cpu_to_le64(con->msgr->supported_features);
Sage Weil's avatar
Sage Weil committed
826 827 828 829 830 831
	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;

832 833
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
834 835
	if (IS_ERR(auth))
		return PTR_ERR(auth);
836

837
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
838 839 840
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

841
	con_out_kvec_add(con, sizeof (con->out_connect),
842 843
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
844
		con_out_kvec_add(con, auth->authorizer_buf_len,
845
					auth->authorizer_buf);
846

Sage Weil's avatar
Sage Weil committed
847
	con->out_more = 0;
848
	set_bit(WRITE_PENDING, &con->flags);
849

850
	return 0;
Sage Weil's avatar
Sage Weil committed
851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872
}

/*
 * 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 */
873 874 875 876 877 878 879 880 881 882 883 884

		/* 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;
Sage Weil's avatar
Sage Weil committed
885 886 887 888 889 890 891 892 893 894 895
		}
	}
	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! */
}

896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922
static void out_msg_pos_next(struct ceph_connection *con, struct page *page,
			size_t len, size_t sent, bool in_trail)
{
	struct ceph_msg *msg = con->out_msg;

	BUG_ON(!msg);
	BUG_ON(!sent);

	con->out_msg_pos.data_pos += sent;
	con->out_msg_pos.page_pos += sent;
	if (sent == len) {
		con->out_msg_pos.page_pos = 0;
		con->out_msg_pos.page++;
		con->out_msg_pos.did_page_crc = false;
		if (in_trail)
			list_move_tail(&page->lru,
				       &msg->trail->head);
		else if (msg->pagelist)
			list_move_tail(&page->lru,
				       &msg->pagelist->head);
#ifdef CONFIG_BLOCK
		else if (msg->bio)
			iter_bio_next(&msg->bio_iter, &msg->bio_seg);
#endif
	}
}

Sage Weil's avatar
Sage Weil committed
923 924 925 926 927 928 929 930 931 932
/*
 * 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;
933
	unsigned int data_len = le32_to_cpu(msg->hdr.data_len);
Sage Weil's avatar
Sage Weil committed
934
	size_t len;
935
	bool do_datacrc = !con->msgr->nocrc;
Sage Weil's avatar
Sage Weil committed
936
	int ret;
937
	int total_max_write;
938
	bool in_trail = false;
939
	size_t trail_len = (msg->trail ? msg->trail->length : 0);
Sage Weil's avatar
Sage Weil committed
940 941

	dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
942
	     con, msg, con->out_msg_pos.page, msg->nr_pages,
Sage Weil's avatar
Sage Weil committed
943 944
	     con->out_msg_pos.page_pos);

945 946 947 948 949 950
#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) {
Sage Weil's avatar
Sage Weil committed
951
		struct page *page = NULL;
952
		int max_write = PAGE_SIZE;
953
		int bio_offset = 0;
954 955 956

		total_max_write = data_len - trail_len -
			con->out_msg_pos.data_pos;
Sage Weil's avatar
Sage Weil committed
957 958 959 960 961 962

		/*
		 * 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.
		 */
963 964 965

		/* have we reached the trail part of the data? */
		if (con->out_msg_pos.data_pos >= data_len - trail_len) {
966
			in_trail = true;
967 968 969 970 971 972

			total_max_write = data_len - con->out_msg_pos.data_pos;

			page = list_first_entry(&msg->trail->head,
						struct page, lru);
		} else if (msg->pages) {
Sage Weil's avatar
Sage Weil committed
973
			page = msg->pages[con->out_msg_pos.page];
974 975 976
		} else if (msg->pagelist) {
			page = list_first_entry(&msg->pagelist->head,
						struct page, lru);
977 978 979 980 981 982
#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;
983
			bio_offset = bv->bv_offset;
984 985
			max_write = bv->bv_len;
#endif
Sage Weil's avatar
Sage Weil committed
986
		} else {
987
			page = zero_page;
Sage Weil's avatar
Sage Weil committed
988
		}
989 990 991
		len = min_t(int, max_write - con->out_msg_pos.page_pos,
			    total_max_write);

992
		if (do_datacrc && !con->out_msg_pos.did_page_crc) {
993
			void *base;
994
			u32 crc;
995
			u32 tmpcrc = le32_to_cpu(msg->footer.data_crc);
996
			char *kaddr;
Sage Weil's avatar
Sage Weil committed
997

998
			kaddr = kmap(page);
Sage Weil's avatar
Sage Weil committed
999
			BUG_ON(kaddr == NULL);
1000
			base = kaddr + con->out_msg_pos.page_pos + bio_offset;
1001
			crc = crc32c(tmpcrc, base, len);
1002
			msg->footer.data_crc = cpu_to_le32(crc);
1003
			con->out_msg_pos.did_page_crc = true;
Sage Weil's avatar
Sage Weil committed
1004
		}
1005
		ret = ceph_tcp_sendpage(con->sock, page,
1006
				      con->out_msg_pos.page_pos + bio_offset,
1007
				      len, 1);
Sage Weil's avatar
Sage Weil committed
1008

1009
		if (do_datacrc)
Sage Weil's avatar
Sage Weil committed
1010 1011 1012 1013 1014
			kunmap(page);

		if (ret <= 0)
			goto out;

1015
		out_msg_pos_next(con, page, len, (size_t) ret, in_trail);
Sage Weil's avatar
Sage Weil committed
1016 1017 1018 1019 1020
	}

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

	/* prepare and queue up footer, too */
1021
	if (!do_datacrc)
1022
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1023
	con_out_kvec_reset(con);
1024
	prepare_write_message_footer(con);
Sage Weil's avatar
Sage Weil committed
1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037
	ret = 1;
out:
	return ret;
}

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

	while (con->out_skip > 0) {
1038
		size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
Sage Weil's avatar
Sage Weil committed
1039

1040
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, 1);
Sage Weil's avatar
Sage Weil committed
1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1053 1054 1055 1056 1057 1058
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

Sage Weil's avatar
Sage Weil committed
1059 1060 1061 1062 1063 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
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,
1092
			int end, int size, void *object)
Sage Weil's avatar
Sage Weil committed
1093
{
1094 1095
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
Sage Weil's avatar
Sage Weil committed
1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108
		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
 */
1109
static int read_partial_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1110
{
1111 1112 1113
	int size;
	int end;
	int ret;
Sage Weil's avatar
Sage Weil committed
1114

1115
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
Sage Weil's avatar
Sage Weil committed
1116 1117

	/* peer's banner */
1118 1119 1120
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
Sage Weil's avatar
Sage Weil committed
1121 1122
	if (ret <= 0)
		goto out;
1123 1124 1125 1126

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
Sage Weil's avatar
Sage Weil committed
1127 1128
	if (ret <= 0)
		goto out;
1129 1130 1131 1132

	size = sizeof (con->peer_addr_for_me);
	end += size;
	ret = read_partial(con, end, size, &con->peer_addr_for_me);
Sage Weil's avatar
Sage Weil committed
1133 1134
	if (ret <= 0)
		goto out;
1135

1136 1137 1138 1139 1140 1141
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1142 1143 1144
	int size;
	int end;
	int ret;
1145 1146 1147

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

1148 1149 1150
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
Sage Weil's avatar
Sage Weil committed
1151 1152
	if (ret <= 0)
		goto out;
1153 1154 1155 1156

	size = le32_to_cpu(con->in_reply.authorizer_len);
	end += size;
	ret = read_partial(con, end, size, con->auth_reply_buf);
1157 1158
	if (ret <= 0)
		goto out;
Sage Weil's avatar
Sage Weil committed
1159

1160 1161 1162
	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),
Sage Weil's avatar
Sage Weil committed
1163 1164 1165
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1166

Sage Weil's avatar
Sage Weil committed
1167 1168 1169 1170 1171 1172 1173 1174
}

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

Noah Watkins's avatar
Noah Watkins committed
1221 1222 1223 1224 1225 1226
/*
 * 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)
{
1227 1228
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
Noah Watkins's avatar
Noah Watkins committed
1229 1230 1231 1232 1233 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

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

Sage Weil's avatar
Sage Weil committed
1316 1317 1318 1319 1320 1321 1322 1323
/*
 * 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)
{
Noah Watkins's avatar
Noah Watkins committed
1324
	int i, ret = -EINVAL;
Sage Weil's avatar
Sage Weil committed
1325 1326 1327 1328 1329 1330 1331
	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;
1332 1333 1334 1335 1336 1337
		char delim = ',';

		if (*p == '[') {
			delim = ']';
			p++;
		}
Sage Weil's avatar
Sage Weil committed
1338

Noah Watkins's avatar
Noah Watkins committed
1339 1340
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
Sage Weil's avatar
Sage Weil committed
1341
			goto bad;
Noah Watkins's avatar
Noah Watkins committed
1342 1343
		ret = -EINVAL;

Sage Weil's avatar
Sage Weil committed
1344 1345
		p = ipend;

1346 1347 1348 1349 1350 1351 1352 1353
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

Sage Weil's avatar
Sage Weil committed
1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369
		/* 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);

1370
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
Sage Weil's avatar
Sage Weil committed
1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386

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

	if (p != end)
		goto bad;

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

bad:
1387
	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
Noah Watkins's avatar
Noah Watkins committed
1388
	return ret;
Sage Weil's avatar
Sage Weil committed
1389
}
1390
EXPORT_SYMBOL(ceph_parse_ips);
Sage Weil's avatar
Sage Weil committed
1391

1392
static int process_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1393
{
1394
	dout("process_banner on %p\n", con);
Sage Weil's avatar
Sage Weil committed
1395 1396 1397 1398

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

1399 1400 1401
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

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

1435 1436 1437 1438 1439
	set_bit(NEGOTIATING, &con->state);
	prepare_read_connect(con);
	return 0;
}

1440 1441 1442 1443 1444 1445
static void fail_protocol(struct ceph_connection *con)
{
	reset_connection(con);
	set_bit(CLOSED, &con->state);  /* in case there's queued work */
}

1446 1447
static int process_connect(struct ceph_connection *con)
{
1448 1449
	u64 sup_feat = con->msgr->supported_features;
	u64 req_feat = con->msgr->required_features;
1450
	u64 server_feat = le64_to_cpu(con->in_reply.features);
1451
	int ret;
1452

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

Sage Weil's avatar
Sage Weil committed
1455
	switch (con->in_reply.tag) {
1456 1457 1458 1459
	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),
1460
		       ceph_pr_addr(&con->peer_addr.in_addr),
1461 1462 1463 1464 1465
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
		fail_protocol(con);
		return -1;

Sage Weil's avatar
Sage Weil committed
1466 1467 1468 1469
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
1470
		       ceph_pr_addr(&con->peer_addr.in_addr),
Sage Weil's avatar
Sage Weil committed
1471 1472 1473
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
1474
		fail_protocol(con);
Sage Weil's avatar
Sage Weil committed
1475 1476
		return -1;

1477 1478 1479 1480 1481 1482 1483 1484 1485
	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;
1486
		con_out_kvec_reset(con);
1487
		ret = prepare_write_connect(con);
1488 1489
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
1490
		prepare_read_connect(con);
1491
		break;
Sage Weil's avatar
Sage Weil committed
1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504

	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),
1505
		       ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
1506
		reset_connection(con);
1507
		con_out_kvec_reset(con);
1508 1509 1510
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
1511 1512 1513
		prepare_read_connect(con);

		/* Tell ceph about it. */
1514
		mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
1515 1516 1517
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
1518
		mutex_lock(&con->mutex);
1519 1520 1521
		if (test_bit(CLOSED, &con->state) ||
		    test_bit(OPENING, &con->state))
			return -EAGAIN;
Sage Weil's avatar
Sage Weil committed
1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532
		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);
1533
		con_out_kvec_reset(con);
1534 1535 1536
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
1537 1538 1539 1540 1541 1542 1543 1544
		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.
		 */
1545
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
Sage Weil's avatar
Sage Weil committed
1546 1547 1548 1549
		     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));
1550
		con_out_kvec_reset(con);
1551 1552 1553
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
1554 1555 1556 1557
		prepare_read_connect(con);
		break;

	case CEPH_MSGR_TAG_READY:
1558 1559 1560 1561
		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),
1562
			       ceph_pr_addr(&con->peer_addr.in_addr),
1563 1564 1565 1566 1567
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
			fail_protocol(con);
			return -1;
		}
Sage Weil's avatar
Sage Weil committed
1568 1569 1570
		clear_bit(CONNECTING, &con->state);
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
1571
		con->peer_features = server_feat;
Sage Weil's avatar
Sage Weil committed
1572 1573 1574 1575 1576 1577
		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));
1578 1579

		if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
1580
			set_bit(LOSSYTX, &con->flags);
1581

Sage Weil's avatar
Sage Weil committed
1582 1583 1584 1585 1586 1587 1588 1589 1590 1591
		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.
		 */
Sage Weil's avatar
Sage Weil committed
1592 1593 1594
		pr_err("process_connect got WAIT as client\n");
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
Sage Weil's avatar
Sage Weil committed
1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609

	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)
{
1610 1611 1612 1613
	int size = sizeof (con->in_temp_ack);
	int end = size;

	return read_partial(con, end, size, &con->in_temp_ack);
Sage Weil's avatar
Sage Weil committed
1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633
}


/*
 * 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);
1634
		m->ack_stamp = jiffies;
Sage Weil's avatar
Sage Weil committed
1635 1636 1637 1638 1639 1640 1641 1642
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}




1643
static int read_partial_message_section(struct ceph_connection *con,
Sage Weil's avatar
Sage Weil committed
1644 1645
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
1646
{
1647
	int ret, left;
1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659

	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;
	}
1660 1661
	if (section->iov_len == sec_len)
		*crc = crc32c(0, section->iov_base, section->iov_len);
Sage Weil's avatar
Sage Weil committed
1662

1663 1664
	return 1;
}
Sage Weil's avatar
Sage Weil committed
1665

Alex Elder's avatar
Alex Elder committed
1666 1667
static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
				struct ceph_msg_header *hdr);
1668 1669 1670 1671


static int read_partial_message_pages(struct ceph_connection *con,
				      struct page **pages,
1672
				      unsigned int data_len, bool do_datacrc)
1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684
{
	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);
1685
	if (ret > 0 && do_datacrc)
1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704
		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,
1705
				    unsigned int data_len, bool do_datacrc)
1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720
{
	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);
1721
	if (ret > 0 && do_datacrc)
1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738
		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

Sage Weil's avatar
Sage Weil committed
1739 1740 1741 1742 1743 1744
/*
 * read (part of) a message.
 */
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
1745 1746
	int size;
	int end;
Sage Weil's avatar
Sage Weil committed
1747
	int ret;
1748
	unsigned int front_len, middle_len, data_len;
1749
	bool do_datacrc = !con->msgr->nocrc;
1750
	u64 seq;
1751
	u32 crc;
Sage Weil's avatar
Sage Weil committed
1752 1753 1754 1755

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

	/* header */
1756 1757 1758
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
1759 1760
	if (ret <= 0)
		return ret;
1761 1762 1763 1764 1765 1766 1767 1768 1769

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

Sage Weil's avatar
Sage Weil committed
1770 1771 1772 1773 1774 1775 1776 1777 1778 1779
	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;

1780 1781 1782
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
1783
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1784
			ENTITY_NAME(con->peer_name),
1785
			ceph_pr_addr(&con->peer_addr.in_addr),
1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797
			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;
	}

Sage Weil's avatar
Sage Weil committed
1798 1799 1800 1801
	/* 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);
Alex Elder's avatar
Alex Elder committed
1802
		if (ceph_con_in_msg_alloc(con, &con->in_hdr)) {
Sage Weil's avatar
Sage Weil committed
1803
			/* skip this message */
1804
			dout("alloc_msg said skip message\n");
1805
			BUG_ON(con->in_msg);
Sage Weil's avatar
Sage Weil committed
1806 1807 1808
			con->in_base_pos = -front_len - middle_len - data_len -
				sizeof(m->footer);
			con->in_tag = CEPH_MSGR_TAG_READY;
1809
			con->in_seq++;
Sage Weil's avatar
Sage Weil committed
1810 1811
			return 0;
		}
1812
		if (!con->in_msg) {
1813 1814
			con->error_msg =
				"error allocating memory for incoming message";
1815
			return -ENOMEM;
Sage Weil's avatar
Sage Weil committed
1816
		}
1817 1818

		BUG_ON(con->in_msg->con != con);
Sage Weil's avatar
Sage Weil committed
1819 1820
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
1821 1822
		if (m->middle)
			m->middle->vec.iov_len = 0;
1823 1824

		con->in_msg_pos.page = 0;
1825
		if (m->pages)
1826
			con->in_msg_pos.page_pos = m->page_alignment;
1827 1828
		else
			con->in_msg_pos.page_pos = 0;
1829
		con->in_msg_pos.data_pos = 0;
Sage Weil's avatar
Sage Weil committed
1830 1831 1832
	}

	/* front */
1833 1834 1835 1836
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
Sage Weil's avatar
Sage Weil committed
1837 1838

	/* middle */
1839
	if (m->middle) {
Sage Weil's avatar
Sage Weil committed
1840 1841
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
1842
						   &con->in_middle_crc);
Sage Weil's avatar
Sage Weil committed
1843 1844 1845
		if (ret <= 0)
			return ret;
	}
1846 1847 1848 1849
#ifdef CONFIG_BLOCK
	if (m->bio && !m->bio_iter)
		init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
#endif
Sage Weil's avatar
Sage Weil committed
1850 1851 1852

	/* (page) data */
	while (con->in_msg_pos.data_pos < data_len) {
1853 1854
		if (m->pages) {
			ret = read_partial_message_pages(con, m->pages,
1855
						 data_len, do_datacrc);
1856 1857 1858 1859 1860 1861 1862
			if (ret <= 0)
				return ret;
#ifdef CONFIG_BLOCK
		} else if (m->bio) {

			ret = read_partial_message_bio(con,
						 &m->bio_iter, &m->bio_seg,
1863
						 data_len, do_datacrc);
1864 1865 1866 1867 1868
			if (ret <= 0)
				return ret;
#endif
		} else {
			BUG_ON(1);
Sage Weil's avatar
Sage Weil committed
1869 1870 1871 1872
		}
	}

	/* footer */
1873 1874 1875
	size = sizeof (m->footer);
	end += size;
	ret = read_partial(con, end, size, &m->footer);
1876 1877 1878
	if (ret <= 0)
		return ret;

Sage Weil's avatar
Sage Weil committed
1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893
	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;
	}
1894
	if (do_datacrc &&
Sage Weil's avatar
Sage Weil committed
1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911
	    (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)
{
Sage Weil's avatar
Sage Weil committed
1912
	struct ceph_msg *msg;
Sage Weil's avatar
Sage Weil committed
1913

1914 1915
	BUG_ON(con->in_msg->con != con);
	con->in_msg->con = NULL;
Sage Weil's avatar
Sage Weil committed
1916
	msg = con->in_msg;
Sage Weil's avatar
Sage Weil committed
1917
	con->in_msg = NULL;
Sage Weil's avatar
Sage Weil committed
1918
	con->ops->put(con);
Sage Weil's avatar
Sage Weil committed
1919 1920 1921

	/* if first message, set peer_name */
	if (con->peer_name.type == 0)
1922
		con->peer_name = msg->hdr.src;
Sage Weil's avatar
Sage Weil committed
1923 1924

	con->in_seq++;
1925
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
1926 1927 1928

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
1929
	     ENTITY_NAME(msg->hdr.src),
Sage Weil's avatar
Sage Weil committed
1930 1931 1932 1933 1934 1935
	     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);
1936 1937

	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949
	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;

1950
	dout("try_write start %p state %lu\n", con, con->state);
Sage Weil's avatar
Sage Weil committed
1951 1952 1953 1954 1955 1956

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

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

1960
		con_out_kvec_reset(con);
1961
		prepare_write_banner(con);
1962 1963 1964
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
1965
		prepare_read_banner(con);
Sage Weil's avatar
Sage Weil committed
1966

1967
		BUG_ON(con->in_msg);
Sage Weil's avatar
Sage Weil committed
1968 1969 1970
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
1971 1972
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
Sage Weil's avatar
Sage Weil committed
1973 1974 1975 1976 1977 1978 1979 1980 1981 1982
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
	if (con->out_skip) {
		ret = write_partial_skip(con);
		if (ret <= 0)
1983
			goto out;
Sage Weil's avatar
Sage Weil committed
1984 1985 1986 1987
	}
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
		if (ret <= 0)
1988
			goto out;
Sage Weil's avatar
Sage Weil committed
1989 1990 1991 1992
	}

	/* msg pages? */
	if (con->out_msg) {
1993 1994 1995 1996 1997 1998
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

Sage Weil's avatar
Sage Weil committed
1999 2000 2001 2002
		ret = write_partial_msg_pages(con);
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2003
			goto out;
Sage Weil's avatar
Sage Weil committed
2004 2005 2006
		if (ret < 0) {
			dout("try_write write_partial_msg_pages err %d\n",
			     ret);
2007
			goto out;
Sage Weil's avatar
Sage Weil committed
2008 2009 2010
		}
	}

2011
do_next:
Sage Weil's avatar
Sage Weil committed
2012 2013 2014 2015 2016 2017 2018 2019 2020 2021
	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;
		}
2022
		if (test_and_clear_bit(KEEPALIVE_PENDING, &con->flags)) {
Sage Weil's avatar
Sage Weil committed
2023 2024 2025 2026 2027 2028
			prepare_write_keepalive(con);
			goto more;
		}
	}

	/* Nothing to do! */
2029
	clear_bit(WRITE_PENDING, &con->flags);
Sage Weil's avatar
Sage Weil committed
2030 2031 2032
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2033
	dout("try_write done on %p ret %d\n", con, ret);
Sage Weil's avatar
Sage Weil committed
2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052
	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);
2053

Sage Weil's avatar
Sage Weil committed
2054 2055 2056
more:
	dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
	     con->in_base_pos);
2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067

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

Sage Weil's avatar
Sage Weil committed
2068
	if (test_bit(CONNECTING, &con->state)) {
2069 2070 2071 2072 2073
		if (!test_bit(NEGOTIATING, &con->state)) {
			dout("try_read connecting\n");
			ret = read_partial_banner(con);
			if (ret <= 0)
				goto out;
2074 2075 2076
			ret = process_banner(con);
			if (ret < 0)
				goto out;
2077
		}
Sage Weil's avatar
Sage Weil committed
2078 2079 2080
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2081 2082 2083
		ret = process_connect(con);
		if (ret < 0)
			goto out;
Sage Weil's avatar
Sage Weil committed
2084 2085 2086 2087 2088 2089 2090 2091 2092
		goto more;
	}

	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
Alex Elder's avatar
Alex Elder committed
2093 2094 2095
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

Sage Weil's avatar
Sage Weil committed
2096 2097 2098
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2099
			goto out;
Sage Weil's avatar
Sage Weil committed
2100 2101 2102 2103 2104 2105 2106 2107 2108 2109
		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)
2110
			goto out;
Sage Weil's avatar
Sage Weil committed
2111 2112 2113 2114 2115 2116 2117 2118 2119 2120
		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 */
2121
			goto out;
Sage Weil's avatar
Sage Weil committed
2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132
		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;
2133
				break;
Sage Weil's avatar
Sage Weil committed
2134 2135
			case -EIO:
				con->error_msg = "io error";
2136
				break;
Sage Weil's avatar
Sage Weil committed
2137
			}
2138
			goto out;
Sage Weil's avatar
Sage Weil committed
2139 2140 2141 2142 2143 2144 2145 2146 2147
		}
		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)
2148
			goto out;
Sage Weil's avatar
Sage Weil committed
2149 2150 2151 2152 2153
		process_ack(con);
		goto more;
	}

out:
2154
	dout("try_read done on %p ret %d\n", con, ret);
Sage Weil's avatar
Sage Weil committed
2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175
	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;
	}

2176
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
Sage Weil's avatar
Sage Weil committed
2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190
		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);
2191
	int ret;
Sage Weil's avatar
Sage Weil committed
2192

Sage Weil's avatar
Sage Weil committed
2193
	mutex_lock(&con->mutex);
2194
restart:
2195
	if (test_and_clear_bit(BACKOFF, &con->flags)) {
Sage Weil's avatar
Sage Weil committed
2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207
		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);
		}
	}
Sage Weil's avatar
Sage Weil committed
2208

Sage Weil's avatar
Sage Weil committed
2209 2210 2211 2212
	if (test_bit(STANDBY, &con->state)) {
		dout("con_work %p STANDBY\n", con);
		goto done;
	}
Sage Weil's avatar
Sage Weil committed
2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223
	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);
	}

2224
	if (test_and_clear_bit(SOCK_CLOSED, &con->flags))
2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237
		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;
Sage Weil's avatar
Sage Weil committed
2238 2239

done:
Sage Weil's avatar
Sage Weil committed
2240 2241
	mutex_unlock(&con->mutex);
done_unlocked:
Sage Weil's avatar
Sage Weil committed
2242
	con->ops->put(con);
2243 2244 2245 2246 2247 2248
	return;

fault:
	mutex_unlock(&con->mutex);
	ceph_fault(con);     /* error/fault path */
	goto done_unlocked;
Sage Weil's avatar
Sage Weil committed
2249 2250 2251 2252 2253 2254 2255 2256 2257 2258
}


/*
 * 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),
2259
	       ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
Sage Weil's avatar
Sage Weil committed
2260
	dout("fault %p state %lu to peer %s\n",
2261
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
2262

2263
	if (test_bit(LOSSYTX, &con->flags)) {
Sage Weil's avatar
Sage Weil committed
2264 2265 2266 2267
		dout("fault on LOSSYTX channel\n");
		goto out;
	}

2268
	mutex_lock(&con->mutex);
2269 2270
	if (test_bit(CLOSED, &con->state))
		goto out_unlock;
2271

Sage Weil's avatar
Sage Weil committed
2272
	con_close_socket(con);
Sage Weil's avatar
Sage Weil committed
2273 2274

	if (con->in_msg) {
2275 2276
		BUG_ON(con->in_msg->con != con);
		con->in_msg->con = NULL;
Sage Weil's avatar
Sage Weil committed
2277 2278
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
Sage Weil's avatar
Sage Weil committed
2279
		con->ops->put(con);
Sage Weil's avatar
Sage Weil committed
2280
	}
Sage Weil's avatar
Sage Weil committed
2281

2282 2283
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2284

Sage Weil's avatar
Sage Weil committed
2285 2286 2287
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2288
	    !test_bit(KEEPALIVE_PENDING, &con->flags)) {
Sage Weil's avatar
Sage Weil committed
2289
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2290
		clear_bit(WRITE_PENDING, &con->flags);
Sage Weil's avatar
Sage Weil committed
2291
		set_bit(STANDBY, &con->state);
2292 2293 2294 2295 2296 2297 2298 2299
	} 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,
Sage Weil's avatar
Sage Weil committed
2300 2301 2302
				       round_jiffies_relative(con->delay))) {
			dout("fault queued %p delay %lu\n", con, con->delay);
		} else {
2303
			con->ops->put(con);
Sage Weil's avatar
Sage Weil committed
2304 2305 2306 2307 2308 2309 2310 2311 2312 2313
			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.
			 */
2314
			set_bit(BACKOFF, &con->flags);
Sage Weil's avatar
Sage Weil committed
2315
		}
Sage Weil's avatar
Sage Weil committed
2316 2317
	}

2318 2319
out_unlock:
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2320
out:
2321 2322 2323
	/*
	 * in case we faulted due to authentication, invalidate our
	 * current tickets so that we can get new ones.
Sage Weil's avatar
Sage Weil committed
2324
	 */
2325 2326 2327 2328 2329
	if (con->auth_retry && con->ops->invalidate_authorizer) {
		dout("calling invalidate_authorizer()\n");
		con->ops->invalidate_authorizer(con);
	}

Sage Weil's avatar
Sage Weil committed
2330 2331 2332 2333 2334 2335 2336
	if (con->ops->fault)
		con->ops->fault(con);
}



/*
2337
 * initialize a new messenger instance
Sage Weil's avatar
Sage Weil committed
2338
 */
2339 2340 2341 2342 2343
void ceph_messenger_init(struct ceph_messenger *msgr,
			struct ceph_entity_addr *myaddr,
			u32 supported_features,
			u32 required_features,
			bool nocrc)
Sage Weil's avatar
Sage Weil committed
2344
{
2345 2346 2347
	msgr->supported_features = supported_features;
	msgr->required_features = required_features;

Sage Weil's avatar
Sage Weil committed
2348 2349 2350 2351 2352 2353
	spin_lock_init(&msgr->global_seq_lock);

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

	/* select a random nonce */
2354
	msgr->inst.addr.type = 0;
Sage Weil's avatar
Sage Weil committed
2355
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2356
	encode_my_addr(msgr);
2357
	msgr->nocrc = nocrc;
Sage Weil's avatar
Sage Weil committed
2358

2359
	dout("%s %p\n", __func__, msgr);
Sage Weil's avatar
Sage Weil committed
2360
}
2361
EXPORT_SYMBOL(ceph_messenger_init);
Sage Weil's avatar
Sage Weil committed
2362

Sage Weil's avatar
Sage Weil committed
2363 2364 2365 2366 2367 2368 2369
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++;
2370 2371
		WARN_ON(test_bit(WRITE_PENDING, &con->flags));
		WARN_ON(test_bit(KEEPALIVE_PENDING, &con->flags));
Sage Weil's avatar
Sage Weil committed
2372 2373 2374 2375
		mutex_unlock(&con->mutex);
	}
}

Sage Weil's avatar
Sage Weil committed
2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387
/*
 * 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 */
2388
	msg->hdr.src = con->msgr->inst.name;
Sage Weil's avatar
Sage Weil committed
2389

2390 2391
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));

2392 2393
	msg->needs_out_seq = true;

Sage Weil's avatar
Sage Weil committed
2394
	/* queue */
2395
	mutex_lock(&con->mutex);
2396

2397
	BUG_ON(msg->con != NULL);
Sage Weil's avatar
Sage Weil committed
2398
	msg->con = con->ops->get(con);
2399 2400
	BUG_ON(msg->con == NULL);

Sage Weil's avatar
Sage Weil committed
2401 2402 2403 2404 2405 2406 2407 2408
	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));
2409
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2410 2411 2412

	/* if there wasn't anything waiting to send before, queue
	 * new work */
Sage Weil's avatar
Sage Weil committed
2413
	clear_standby(con);
2414
	if (test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
Sage Weil's avatar
Sage Weil committed
2415 2416
		queue_con(con);
}
2417
EXPORT_SYMBOL(ceph_con_send);
Sage Weil's avatar
Sage Weil committed
2418 2419 2420 2421

/*
 * Revoke a message that was previously queued for send
 */
2422
void ceph_msg_revoke(struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
2423
{
2424 2425 2426 2427 2428
	struct ceph_connection *con = msg->con;

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

2429
	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2430
	if (!list_empty(&msg->list_head)) {
2431
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
Sage Weil's avatar
Sage Weil committed
2432
		list_del_init(&msg->list_head);
2433
		BUG_ON(msg->con == NULL);
Sage Weil's avatar
Sage Weil committed
2434
		msg->con->ops->put(msg->con);
2435
		msg->con = NULL;
2436
		msg->hdr.seq = 0;
2437

Sage Weil's avatar
Sage Weil committed
2438
		ceph_msg_put(msg);
Sage Weil's avatar
Sage Weil committed
2439 2440
	}
	if (con->out_msg == msg) {
2441
		dout("%s %p msg %p - was sending\n", __func__, con, msg);
Sage Weil's avatar
Sage Weil committed
2442
		con->out_msg = NULL;
Sage Weil's avatar
Sage Weil committed
2443 2444 2445 2446
		if (con->out_kvec_is_msg) {
			con->out_skip = con->out_kvec_bytes;
			con->out_kvec_is_msg = false;
		}
Sage Weil's avatar
Sage Weil committed
2447
		msg->hdr.seq = 0;
2448 2449

		ceph_msg_put(msg);
Sage Weil's avatar
Sage Weil committed
2450
	}
2451
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2452 2453
}

2454
/*
2455
 * Revoke a message that we may be reading data into
2456
 */
2457
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2458
{
2459 2460 2461 2462 2463 2464 2465 2466 2467 2468
	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;
2469
	mutex_lock(&con->mutex);
2470
	if (con->in_msg == msg) {
2471 2472 2473
		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);
2474 2475

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

Sage Weil's avatar
Sage Weil committed
2494 2495 2496 2497 2498
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
Sage Weil's avatar
Sage Weil committed
2499 2500
	dout("con_keepalive %p\n", con);
	clear_standby(con);
2501 2502
	if (test_and_set_bit(KEEPALIVE_PENDING, &con->flags) == 0 &&
	    test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
Sage Weil's avatar
Sage Weil committed
2503 2504
		queue_con(con);
}
2505
EXPORT_SYMBOL(ceph_con_keepalive);
Sage Weil's avatar
Sage Weil committed
2506 2507 2508 2509 2510 2511


/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
2512 2513
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
Sage Weil's avatar
Sage Weil committed
2514 2515 2516
{
	struct ceph_msg *m;

2517
	m = kmalloc(sizeof(*m), flags);
Sage Weil's avatar
Sage Weil committed
2518 2519
	if (m == NULL)
		goto out;
Sage Weil's avatar
Sage Weil committed
2520
	kref_init(&m->kref);
2521 2522

	m->con = NULL;
Sage Weil's avatar
Sage Weil committed
2523 2524
	INIT_LIST_HEAD(&m->list_head);

2525
	m->hdr.tid = 0;
Sage Weil's avatar
Sage Weil committed
2526
	m->hdr.type = cpu_to_le16(type);
2527 2528
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
	m->hdr.version = 0;
Sage Weil's avatar
Sage Weil committed
2529 2530
	m->hdr.front_len = cpu_to_le32(front_len);
	m->hdr.middle_len = 0;
Sage Weil's avatar
Sage Weil committed
2531 2532
	m->hdr.data_len = 0;
	m->hdr.data_off = 0;
2533
	m->hdr.reserved = 0;
Sage Weil's avatar
Sage Weil committed
2534 2535 2536
	m->footer.front_crc = 0;
	m->footer.middle_crc = 0;
	m->footer.data_crc = 0;
2537
	m->footer.flags = 0;
Sage Weil's avatar
Sage Weil committed
2538 2539 2540
	m->front_max = front_len;
	m->front_is_vmalloc = false;
	m->more_to_follow = false;
2541
	m->ack_stamp = 0;
Sage Weil's avatar
Sage Weil committed
2542 2543
	m->pool = NULL;

2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556
	/* 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;

Sage Weil's avatar
Sage Weil committed
2557 2558 2559
	/* front */
	if (front_len) {
		if (front_len > PAGE_CACHE_SIZE) {
2560
			m->front.iov_base = __vmalloc(front_len, flags,
Sage Weil's avatar
Sage Weil committed
2561 2562 2563
						      PAGE_KERNEL);
			m->front_is_vmalloc = true;
		} else {
2564
			m->front.iov_base = kmalloc(front_len, flags);
Sage Weil's avatar
Sage Weil committed
2565 2566
		}
		if (m->front.iov_base == NULL) {
2567
			dout("ceph_msg_new can't allocate %d bytes\n",
Sage Weil's avatar
Sage Weil committed
2568 2569 2570 2571 2572 2573 2574 2575
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
	m->front.iov_len = front_len;

Sage Weil's avatar
Sage Weil committed
2576
	dout("ceph_msg_new %p front %d\n", m, front_len);
Sage Weil's avatar
Sage Weil committed
2577 2578 2579 2580 2581
	return m;

out2:
	ceph_msg_put(m);
out:
2582 2583 2584
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
2585
		WARN_ON(1);
2586 2587 2588 2589
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
2590
	return NULL;
Sage Weil's avatar
Sage Weil committed
2591
}
2592
EXPORT_SYMBOL(ceph_msg_new);
Sage Weil's avatar
Sage Weil committed
2593 2594 2595 2596 2597 2598 2599 2600

/*
 * 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.
 */
2601
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
2602 2603 2604 2605 2606 2607 2608 2609 2610
{
	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);

2611
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
Sage Weil's avatar
Sage Weil committed
2612 2613 2614 2615 2616
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

2617
/*
Alex Elder's avatar
Alex Elder committed
2618 2619 2620 2621 2622 2623 2624 2625
 * 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.
2626
 */
Alex Elder's avatar
Alex Elder committed
2627 2628
static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
				struct ceph_msg_header *hdr)
2629 2630 2631 2632 2633 2634
{
	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;

Alex Elder's avatar
Alex Elder committed
2635 2636
	BUG_ON(con->in_msg != NULL);

2637
	if (con->ops->alloc_msg) {
Alex Elder's avatar
Alex Elder committed
2638 2639
		int skip = 0;

2640
		mutex_unlock(&con->mutex);
Alex Elder's avatar
Alex Elder committed
2641
		con->in_msg = con->ops->alloc_msg(con, hdr, &skip);
2642
		mutex_lock(&con->mutex);
2643
		if (con->in_msg) {
Sage Weil's avatar
Sage Weil committed
2644
			con->in_msg->con = con->ops->get(con);
2645 2646
			BUG_ON(con->in_msg->con == NULL);
		}
Alex Elder's avatar
Alex Elder committed
2647 2648 2649 2650 2651
		if (skip)
			con->in_msg = NULL;

		if (!con->in_msg)
			return skip != 0;
2652
	}
Alex Elder's avatar
Alex Elder committed
2653 2654 2655
	if (!con->in_msg) {
		con->in_msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
		if (!con->in_msg) {
2656 2657
			pr_err("unable to allocate msg type %d len %d\n",
			       type, front_len);
Alex Elder's avatar
Alex Elder committed
2658
			return false;
2659
		}
Sage Weil's avatar
Sage Weil committed
2660
		con->in_msg->con = con->ops->get(con);
2661
		BUG_ON(con->in_msg->con == NULL);
Alex Elder's avatar
Alex Elder committed
2662
		con->in_msg->page_alignment = le16_to_cpu(hdr->data_off);
2663
	}
Alex Elder's avatar
Alex Elder committed
2664
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2665

Alex Elder's avatar
Alex Elder committed
2666 2667
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
2668
		if (ret < 0) {
Alex Elder's avatar
Alex Elder committed
2669 2670
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
2671 2672
		}
	}
2673

Alex Elder's avatar
Alex Elder committed
2674
	return false;
2675 2676
}

Sage Weil's avatar
Sage Weil committed
2677 2678 2679 2680 2681 2682 2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693

/*
 * 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.
 */
Sage Weil's avatar
Sage Weil committed
2694 2695 2696
void ceph_msg_last_put(struct kref *kref)
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
Sage Weil's avatar
Sage Weil committed
2697

Sage Weil's avatar
Sage Weil committed
2698 2699 2700 2701 2702 2703 2704
	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;
Sage Weil's avatar
Sage Weil committed
2705
	}
Sage Weil's avatar
Sage Weil committed
2706 2707 2708
	m->nr_pages = 0;
	m->pages = NULL;

2709 2710 2711 2712 2713 2714
	if (m->pagelist) {
		ceph_pagelist_release(m->pagelist);
		kfree(m->pagelist);
		m->pagelist = NULL;
	}

2715 2716
	m->trail = NULL;

Sage Weil's avatar
Sage Weil committed
2717 2718 2719 2720
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
		ceph_msg_kfree(m);
Sage Weil's avatar
Sage Weil committed
2721
}
2722
EXPORT_SYMBOL(ceph_msg_last_put);
2723 2724 2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742

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);
}
2743
EXPORT_SYMBOL(ceph_msg_dump);