messenger.c 87.9 KB
Newer Older
1
// SPDX-License-Identifier: GPL-2.0
2
#include <linux/ceph/ceph_debug.h>
S
Sage Weil 已提交
3 4 5 6 7 8 9

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

I
Ilya Dryomov 已提交
21
#include <linux/ceph/ceph_features.h>
22 23 24 25
#include <linux/ceph/libceph.h>
#include <linux/ceph/messenger.h>
#include <linux/ceph/decode.h>
#include <linux/ceph/pagelist.h>
26
#include <linux/export.h>
S
Sage Weil 已提交
27 28 29 30 31 32 33 34 35 36

/*
 * Ceph uses the messenger to exchange ceph_msg messages with other
 * hosts in the system.  The messenger provides ordered and reliable
 * delivery.  We tolerate TCP disconnects by reconnecting (with
 * exponential backoff) in the case of a fault (disconnection, bad
 * crc, protocol error).  Acks allow sent messages to be discarded by
 * the sender.
 */

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

S
Sage Weil 已提交
161 162 163 164
/* 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;
165
static char tag_keepalive2 = CEPH_MSGR_TAG_KEEPALIVE2;
S
Sage Weil 已提交
166

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

A
Alex Elder 已提交
171 172 173 174 175
/*
 * When skipping (ignoring) a block of input we read it into a "skip
 * buffer," which is this many bytes in size.
 */
#define SKIP_BUF_SIZE	1024
S
Sage Weil 已提交
176 177

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

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

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

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

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

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

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

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

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

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

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

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

237 238 239
static int ceph_msgr_slab_init(void)
{
	BUG_ON(ceph_msg_cache);
G
Geliang Tang 已提交
240
	ceph_msg_cache = KMEM_CACHE(ceph_msg, 0);
241 242 243 244
	if (!ceph_msg_cache)
		return -ENOMEM;

	BUG_ON(ceph_msg_data_cache);
G
Geliang Tang 已提交
245
	ceph_msg_data_cache = KMEM_CACHE(ceph_msg_data, 0);
246 247 248 249 250 251 252
	if (ceph_msg_data_cache)
		return 0;

	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;

	return -ENOMEM;
253 254 255 256
}

static void ceph_msgr_slab_exit(void)
{
257 258 259 260
	BUG_ON(!ceph_msg_data_cache);
	kmem_cache_destroy(ceph_msg_data_cache);
	ceph_msg_data_cache = NULL;

261 262 263 264 265
	BUG_ON(!ceph_msg_cache);
	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;
}

A
Alex Elder 已提交
266
static void _ceph_msgr_exit(void)
267
{
A
Alex Elder 已提交
268
	if (ceph_msgr_wq) {
269
		destroy_workqueue(ceph_msgr_wq);
A
Alex Elder 已提交
270 271
		ceph_msgr_wq = NULL;
	}
272 273

	BUG_ON(zero_page == NULL);
274
	put_page(zero_page);
275
	zero_page = NULL;
276 277

	ceph_msgr_slab_exit();
278 279
}

280
int ceph_msgr_init(void)
S
Sage Weil 已提交
281
{
282 283 284
	if (ceph_msgr_slab_init())
		return -ENOMEM;

285 286
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
287
	get_page(zero_page);
288

289 290 291 292 293
	/*
	 * The number of active work items is limited by the number of
	 * connections, so leave @max_active at default.
	 */
	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_MEM_RECLAIM, 0);
294 295
	if (ceph_msgr_wq)
		return 0;
296

297 298
	pr_err("msgr_init failed to create workqueue\n");
	_ceph_msgr_exit();
299

300
	return -ENOMEM;
S
Sage Weil 已提交
301
}
302
EXPORT_SYMBOL(ceph_msgr_init);
S
Sage Weil 已提交
303 304 305

void ceph_msgr_exit(void)
{
306 307
	BUG_ON(ceph_msgr_wq == NULL);

308
	_ceph_msgr_exit();
S
Sage Weil 已提交
309
}
310
EXPORT_SYMBOL(ceph_msgr_exit);
S
Sage Weil 已提交
311

Y
Yehuda Sadeh 已提交
312
void ceph_msgr_flush(void)
313 314 315
{
	flush_workqueue(ceph_msgr_wq);
}
316
EXPORT_SYMBOL(ceph_msgr_flush);
317

318 319 320 321 322 323 324 325 326
/* 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);
327 328
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
329 330 331 332 333 334 335 336 337
}

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);
338 339
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTING);
340 341 342 343 344 345 346 347 348
}

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);
349 350
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTED);
351 352 353 354 355 356 357 358 359 360 361
}

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);
362 363
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSING);
364 365 366 367 368 369 370 371
}

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 &&
372
		    old_state != CON_SOCK_STATE_CLOSING &&
373 374
		    old_state != CON_SOCK_STATE_CONNECTING &&
		    old_state != CON_SOCK_STATE_CLOSED))
375
		printk("%s: unexpected old state %d\n", __func__, old_state);
376 377
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
378
}
379

S
Sage Weil 已提交
380 381 382 383 384
/*
 * socket callback functions
 */

/* data available on socket, or listen socket received a connect */
385
static void ceph_sock_data_ready(struct sock *sk)
S
Sage Weil 已提交
386
{
A
Alex Elder 已提交
387
	struct ceph_connection *con = sk->sk_user_data;
388 389 390
	if (atomic_read(&con->msgr->stopping)) {
		return;
	}
A
Alex Elder 已提交
391

S
Sage Weil 已提交
392
	if (sk->sk_state != TCP_CLOSE_WAIT) {
A
Alex Elder 已提交
393
		dout("%s on %p state = %lu, queueing work\n", __func__,
S
Sage Weil 已提交
394 395 396 397 398 399
		     con, con->state);
		queue_con(con);
	}
}

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

404 405
	/* only queue to workqueue if there is data we want to write,
	 * and there is sufficient space in the socket buffer to accept
A
Alex Elder 已提交
406
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
407 408 409 410
	 * 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().
	 */
411
	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
412
		if (sk_stream_is_writeable(sk)) {
A
Alex Elder 已提交
413
			dout("%s %p queueing write work\n", __func__, con);
414 415 416
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
S
Sage Weil 已提交
417
	} else {
A
Alex Elder 已提交
418
		dout("%s %p nothing to write\n", __func__, con);
S
Sage Weil 已提交
419 420 421 422
	}
}

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

A
Alex Elder 已提交
427
	dout("%s %p state = %lu sk_state = %u\n", __func__,
S
Sage Weil 已提交
428 429 430 431
	     con, con->state, sk->sk_state);

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

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
A
Alex Elder 已提交
457
	sk->sk_user_data = con;
A
Alex Elder 已提交
458 459 460
	sk->sk_data_ready = ceph_sock_data_ready;
	sk->sk_write_space = ceph_sock_write_space;
	sk->sk_state_change = ceph_sock_state_change;
S
Sage Weil 已提交
461 462 463 464 465 466 467 468 469 470
}


/*
 * socket helpers
 */

/*
 * initiate connection to a remote socket.
 */
471
static int ceph_tcp_connect(struct ceph_connection *con)
S
Sage Weil 已提交
472
{
S
Sage Weil 已提交
473
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
S
Sage Weil 已提交
474
	struct socket *sock;
475
	unsigned int noio_flag;
S
Sage Weil 已提交
476 477 478
	int ret;

	BUG_ON(con->sock);
479 480 481

	/* sock_create_kern() allocates with GFP_KERNEL */
	noio_flag = memalloc_noio_save();
482
	ret = sock_create_kern(read_pnet(&con->msgr->net), paddr->ss_family,
483
			       SOCK_STREAM, IPPROTO_TCP, &sock);
484
	memalloc_noio_restore(noio_flag);
S
Sage Weil 已提交
485
	if (ret)
486
		return ret;
487
	sock->sk->sk_allocation = GFP_NOFS;
S
Sage Weil 已提交
488

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

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

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

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

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

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

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

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

A
Al Viro 已提交
531 532
	iov_iter_kvec(&msg.msg_iter, READ | ITER_KVEC, &iov, 1, len);
	r = sock_recvmsg(sock, &msg, msg.msg_flags);
533 534 535
	if (r == -EAGAIN)
		r = 0;
	return r;
S
Sage Weil 已提交
536 537
}

538 539 540
static int ceph_tcp_recvpage(struct socket *sock, struct page *page,
		     int page_offset, size_t length)
{
A
Al Viro 已提交
541 542 543 544 545 546 547
	struct bio_vec bvec = {
		.bv_page = page,
		.bv_offset = page_offset,
		.bv_len = length
	};
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
	int r;
548 549

	BUG_ON(page_offset + length > PAGE_SIZE);
A
Al Viro 已提交
550 551 552 553 554
	iov_iter_bvec(&msg.msg_iter, READ | ITER_BVEC, &bvec, 1, length);
	r = sock_recvmsg(sock, &msg, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
555 556
}

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

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

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

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

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

	return ret;
}

591 592 593
static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
		     int offset, size_t size, bool more)
{
594 595
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
	struct bio_vec bvec;
596 597 598 599 600 601 602
	int ret;

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

603 604 605 606 607 608 609 610 611 612 613 614 615
	bvec.bv_page = page;
	bvec.bv_offset = offset;
	bvec.bv_len = size;

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

	iov_iter_bvec(&msg.msg_iter, WRITE | ITER_BVEC, &bvec, 1, size);
	ret = sock_sendmsg(sock, &msg);
	if (ret == -EAGAIN)
		ret = 0;
616 617 618

	return ret;
}
S
Sage Weil 已提交
619 620 621 622 623 624

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

	dout("con_close_socket on %p sock %p\n", con, con->sock);
628 629 630 631 632
	if (con->sock) {
		rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
		sock_release(con->sock);
		con->sock = NULL;
	}
633 634

	/*
S
Sage Weil 已提交
635
	 * Forcibly clear the SOCK_CLOSED flag.  It gets set
636 637 638 639
	 * independent of the connection mutex, and we could have
	 * received a socket close event before we had the chance to
	 * shut the socket down.
	 */
640
	con_flag_clear(con, CON_FLAG_SOCK_CLOSED);
641

642
	con_sock_state_closed(con);
S
Sage Weil 已提交
643 644 645 646 647 648 649 650 651 652
	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);
653

S
Sage Weil 已提交
654 655 656 657 658 659 660 661 662 663 664 665 666 667 668
	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 */
669
	dout("reset_connection %p\n", con);
S
Sage Weil 已提交
670 671 672
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

673
	if (con->in_msg) {
674
		BUG_ON(con->in_msg->con != con);
675 676 677 678
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
	}

S
Sage Weil 已提交
679 680
	con->connect_seq = 0;
	con->out_seq = 0;
681
	if (con->out_msg) {
682
		BUG_ON(con->out_msg->con != con);
683 684 685
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
S
Sage Weil 已提交
686
	con->in_seq = 0;
687
	con->in_seq_acked = 0;
I
Ilya Dryomov 已提交
688 689

	con->out_skip = 0;
S
Sage Weil 已提交
690 691 692 693 694 695 696
}

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

702 703 704 705
	con_flag_clear(con, CON_FLAG_LOSSYTX);	/* so we retry next connect */
	con_flag_clear(con, CON_FLAG_KEEPALIVE_PENDING);
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
	con_flag_clear(con, CON_FLAG_BACKOFF);
706

S
Sage Weil 已提交
707
	reset_connection(con);
S
Sage Weil 已提交
708
	con->peer_global_seq = 0;
709
	cancel_con(con);
710
	con_close_socket(con);
711
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
712
}
713
EXPORT_SYMBOL(ceph_con_close);
S
Sage Weil 已提交
714 715 716 717

/*
 * Reopen a closed connection, with a new peer address.
 */
718 719 720
void ceph_con_open(struct ceph_connection *con,
		   __u8 entity_type, __u64 entity_num,
		   struct ceph_entity_addr *addr)
S
Sage Weil 已提交
721
{
722
	mutex_lock(&con->mutex);
723
	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
724

725
	WARN_ON(con->state != CON_STATE_CLOSED);
726
	con->state = CON_STATE_PREOPEN;
727

728 729 730
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

S
Sage Weil 已提交
731
	memcpy(&con->peer_addr, addr, sizeof(*addr));
732
	con->delay = 0;      /* reset backoff memory */
733
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
734 735
	queue_con(con);
}
736
EXPORT_SYMBOL(ceph_con_open);
S
Sage Weil 已提交
737

738 739 740 741 742 743 744 745
/*
 * return true if this connection ever successfully opened
 */
bool ceph_con_opened(struct ceph_connection *con)
{
	return con->connect_seq > 0;
}

S
Sage Weil 已提交
746 747 748
/*
 * initialize a new connection.
 */
749 750
void ceph_con_init(struct ceph_connection *con, void *private,
	const struct ceph_connection_operations *ops,
751
	struct ceph_messenger *msgr)
S
Sage Weil 已提交
752 753 754
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
755 756
	con->private = private;
	con->ops = ops;
S
Sage Weil 已提交
757
	con->msgr = msgr;
758 759 760

	con_sock_state_init(con);

761
	mutex_init(&con->mutex);
S
Sage Weil 已提交
762 763
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
764
	INIT_DELAYED_WORK(&con->work, ceph_con_workfn);
765

766
	con->state = CON_STATE_CLOSED;
S
Sage Weil 已提交
767
}
768
EXPORT_SYMBOL(ceph_con_init);
S
Sage Weil 已提交
769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786


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

787
static void con_out_kvec_reset(struct ceph_connection *con)
788
{
I
Ilya Dryomov 已提交
789 790
	BUG_ON(con->out_skip);

791 792 793 794 795
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

796
static void con_out_kvec_add(struct ceph_connection *con,
797 798
				size_t size, void *data)
{
I
Ilya Dryomov 已提交
799
	int index = con->out_kvec_left;
800

I
Ilya Dryomov 已提交
801
	BUG_ON(con->out_skip);
802 803 804 805 806 807 808
	BUG_ON(index >= ARRAY_SIZE(con->out_kvec));

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

I
Ilya Dryomov 已提交
810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830
/*
 * Chop off a kvec from the end.  Return residual number of bytes for
 * that kvec, i.e. how many bytes would have been written if the kvec
 * hadn't been nuked.
 */
static int con_out_kvec_skip(struct ceph_connection *con)
{
	int off = con->out_kvec_cur - con->out_kvec;
	int skip = 0;

	if (con->out_kvec_bytes > 0) {
		skip = con->out_kvec[off + con->out_kvec_left - 1].iov_len;
		BUG_ON(con->out_kvec_bytes < skip);
		BUG_ON(!con->out_kvec_left);
		con->out_kvec_bytes -= skip;
		con->out_kvec_left--;
	}

	return skip;
}

831
#ifdef CONFIG_BLOCK
832 833 834 835 836 837

/*
 * For a bio data item, a piece is whatever remains of the next
 * entry in the current bio iovec, or the first entry in the next
 * bio in the list.
 */
A
Alex Elder 已提交
838
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
839
					size_t length)
840
{
A
Alex Elder 已提交
841
	struct ceph_msg_data *data = cursor->data;
842
	struct ceph_bio_iter *it = &cursor->bio_iter;
843

844 845 846 847
	cursor->resid = min_t(size_t, length, data->bio_length);
	*it = data->bio_pos;
	if (cursor->resid < it->iter.bi_size)
		it->iter.bi_size = cursor->resid;
848

849 850
	BUG_ON(cursor->resid < bio_iter_len(it->bio, it->iter));
	cursor->last_piece = cursor->resid == bio_iter_len(it->bio, it->iter);
851 852
}

A
Alex Elder 已提交
853
static struct page *ceph_msg_data_bio_next(struct ceph_msg_data_cursor *cursor,
854 855 856
						size_t *page_offset,
						size_t *length)
{
857 858
	struct bio_vec bv = bio_iter_iovec(cursor->bio_iter.bio,
					   cursor->bio_iter.iter);
859

860 861 862
	*page_offset = bv.bv_offset;
	*length = bv.bv_len;
	return bv.bv_page;
863 864
}

A
Alex Elder 已提交
865 866
static bool ceph_msg_data_bio_advance(struct ceph_msg_data_cursor *cursor,
					size_t bytes)
867
{
868
	struct ceph_bio_iter *it = &cursor->bio_iter;
869

870 871
	BUG_ON(bytes > cursor->resid);
	BUG_ON(bytes > bio_iter_len(it->bio, it->iter));
872
	cursor->resid -= bytes;
873
	bio_advance_iter(it->bio, &it->iter, bytes);
874

875 876 877 878
	if (!cursor->resid) {
		BUG_ON(!cursor->last_piece);
		return false;   /* no more data */
	}
879

880
	if (!bytes || (it->iter.bi_size && it->iter.bi_bvec_done))
881 882
		return false;	/* more bytes to process in this segment */

883 884 885 886 887
	if (!it->iter.bi_size) {
		it->bio = it->bio->bi_next;
		it->iter = it->bio->bi_iter;
		if (cursor->resid < it->iter.bi_size)
			it->iter.bi_size = cursor->resid;
888
	}
889

890 891 892
	BUG_ON(cursor->last_piece);
	BUG_ON(cursor->resid < bio_iter_len(it->bio, it->iter));
	cursor->last_piece = cursor->resid == bio_iter_len(it->bio, it->iter);
893 894
	return true;
}
895
#endif /* CONFIG_BLOCK */
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 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948
static void ceph_msg_data_bvecs_cursor_init(struct ceph_msg_data_cursor *cursor,
					size_t length)
{
	struct ceph_msg_data *data = cursor->data;
	struct bio_vec *bvecs = data->bvec_pos.bvecs;

	cursor->resid = min_t(size_t, length, data->bvec_pos.iter.bi_size);
	cursor->bvec_iter = data->bvec_pos.iter;
	cursor->bvec_iter.bi_size = cursor->resid;

	BUG_ON(cursor->resid < bvec_iter_len(bvecs, cursor->bvec_iter));
	cursor->last_piece =
	    cursor->resid == bvec_iter_len(bvecs, cursor->bvec_iter);
}

static struct page *ceph_msg_data_bvecs_next(struct ceph_msg_data_cursor *cursor,
						size_t *page_offset,
						size_t *length)
{
	struct bio_vec bv = bvec_iter_bvec(cursor->data->bvec_pos.bvecs,
					   cursor->bvec_iter);

	*page_offset = bv.bv_offset;
	*length = bv.bv_len;
	return bv.bv_page;
}

static bool ceph_msg_data_bvecs_advance(struct ceph_msg_data_cursor *cursor,
					size_t bytes)
{
	struct bio_vec *bvecs = cursor->data->bvec_pos.bvecs;

	BUG_ON(bytes > cursor->resid);
	BUG_ON(bytes > bvec_iter_len(bvecs, cursor->bvec_iter));
	cursor->resid -= bytes;
	bvec_iter_advance(bvecs, &cursor->bvec_iter, bytes);

	if (!cursor->resid) {
		BUG_ON(!cursor->last_piece);
		return false;   /* no more data */
	}

	if (!bytes || cursor->bvec_iter.bi_bvec_done)
		return false;	/* more bytes to process in this segment */

	BUG_ON(cursor->last_piece);
	BUG_ON(cursor->resid < bvec_iter_len(bvecs, cursor->bvec_iter));
	cursor->last_piece =
	    cursor->resid == bvec_iter_len(bvecs, cursor->bvec_iter);
	return true;
}

949 950 951 952
/*
 * For a page array, a piece comes from the first page in the array
 * that has not already been fully consumed.
 */
A
Alex Elder 已提交
953
static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor,
954
					size_t length)
955
{
A
Alex Elder 已提交
956
	struct ceph_msg_data *data = cursor->data;
957 958 959 960 961 962 963
	int page_count;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

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

964
	cursor->resid = min(length, data->length);
965 966 967
	page_count = calc_pages_for(data->alignment, (u64)data->length);
	cursor->page_offset = data->alignment & ~PAGE_MASK;
	cursor->page_index = 0;
968 969 970
	BUG_ON(page_count > (int)USHRT_MAX);
	cursor->page_count = (unsigned short)page_count;
	BUG_ON(length > SIZE_MAX - cursor->page_offset);
971
	cursor->last_piece = cursor->page_offset + cursor->resid <= PAGE_SIZE;
972 973
}

A
Alex Elder 已提交
974 975 976
static struct page *
ceph_msg_data_pages_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length)
977
{
A
Alex Elder 已提交
978
	struct ceph_msg_data *data = cursor->data;
979 980 981 982 983 984 985

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

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

	*page_offset = cursor->page_offset;
986
	if (cursor->last_piece)
987
		*length = cursor->resid;
988
	else
989 990 991 992 993
		*length = PAGE_SIZE - *page_offset;

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

A
Alex Elder 已提交
994
static bool ceph_msg_data_pages_advance(struct ceph_msg_data_cursor *cursor,
995 996
						size_t bytes)
{
A
Alex Elder 已提交
997
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_PAGES);
998 999 1000 1001 1002 1003

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

	/* Advance the cursor page offset */

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

1008 1009 1010
	if (!cursor->resid)
		return false;   /* no more data */

1011
	/* Move on to the next page; offset is already at 0 */
1012 1013 1014

	BUG_ON(cursor->page_index >= cursor->page_count);
	cursor->page_index++;
1015
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1016 1017 1018 1019

	return true;
}

1020
/*
1021 1022
 * For a pagelist, a piece is whatever remains to be consumed in the
 * first page in the list, or the front of the next page.
1023
 */
A
Alex Elder 已提交
1024 1025
static void
ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
1026
					size_t length)
1027
{
A
Alex Elder 已提交
1028
	struct ceph_msg_data *data = cursor->data;
1029 1030 1031
	struct ceph_pagelist *pagelist;
	struct page *page;

1032
	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
1033 1034 1035

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

	if (!length)
1038 1039 1040 1041 1042
		return;		/* pagelist can be assigned but empty */

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

1043
	cursor->resid = min(length, pagelist->length);
1044 1045
	cursor->page = page;
	cursor->offset = 0;
A
Alex Elder 已提交
1046
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1047 1048
}

A
Alex Elder 已提交
1049 1050 1051
static struct page *
ceph_msg_data_pagelist_next(struct ceph_msg_data_cursor *cursor,
				size_t *page_offset, size_t *length)
1052
{
A
Alex Elder 已提交
1053
	struct ceph_msg_data *data = cursor->data;
1054 1055 1056 1057 1058 1059 1060 1061
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(!cursor->page);
1062
	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1063

1064
	/* offset of first page in pagelist is always 0 */
1065
	*page_offset = cursor->offset & ~PAGE_MASK;
1066
	if (cursor->last_piece)
1067 1068 1069
		*length = cursor->resid;
	else
		*length = PAGE_SIZE - *page_offset;
1070

A
Alex Elder 已提交
1071
	return cursor->page;
1072 1073
}

A
Alex Elder 已提交
1074
static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor,
1075
						size_t bytes)
1076
{
A
Alex Elder 已提交
1077
	struct ceph_msg_data *data = cursor->data;
1078 1079 1080 1081 1082 1083
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

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

	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1086 1087 1088 1089
	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);

	/* Advance the cursor offset */

1090
	cursor->resid -= bytes;
1091
	cursor->offset += bytes;
1092
	/* offset of first page in pagelist is always 0 */
1093 1094 1095
	if (!bytes || cursor->offset & ~PAGE_MASK)
		return false;	/* more bytes to process in the current page */

1096 1097 1098
	if (!cursor->resid)
		return false;   /* no more data */

1099 1100 1101
	/* Move on to the next page */

	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
1102
	cursor->page = list_next_entry(cursor->page, lru);
1103
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1104 1105 1106 1107

	return true;
}

1108 1109 1110 1111 1112 1113 1114 1115
/*
 * Message data is handled (sent or received) in pieces, where each
 * piece resides on a single page.  The network layer might not
 * consume an entire piece at once.  A data item's cursor keeps
 * track of which piece is next to process and how much remains to
 * be processed in that piece.  It also tracks whether the current
 * piece is the last one in the data item.
 */
1116
static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
1117
{
1118
	size_t length = cursor->total_resid;
A
Alex Elder 已提交
1119 1120

	switch (cursor->data->type) {
1121
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1122
		ceph_msg_data_pagelist_cursor_init(cursor, length);
1123
		break;
1124
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1125
		ceph_msg_data_pages_cursor_init(cursor, length);
1126
		break;
1127 1128
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1129
		ceph_msg_data_bio_cursor_init(cursor, length);
1130
		break;
1131
#endif /* CONFIG_BLOCK */
1132 1133 1134
	case CEPH_MSG_DATA_BVECS:
		ceph_msg_data_bvecs_cursor_init(cursor, length);
		break;
1135
	case CEPH_MSG_DATA_NONE:
1136 1137 1138 1139
	default:
		/* BUG(); */
		break;
	}
A
Alex Elder 已提交
1140
	cursor->need_crc = true;
1141 1142
}

1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159
static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
{
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
	struct ceph_msg_data *data;

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

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

	__ceph_msg_data_cursor_init(cursor);
}

1160 1161 1162 1163 1164
/*
 * Return the page containing the next piece to process for a given
 * data item, and supply the page offset and length of that piece.
 * Indicate whether this is the last piece in this data item.
 */
A
Alex Elder 已提交
1165 1166
static struct page *ceph_msg_data_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length,
1167 1168 1169 1170
					bool *last_piece)
{
	struct page *page;

A
Alex Elder 已提交
1171
	switch (cursor->data->type) {
1172
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1173
		page = ceph_msg_data_pagelist_next(cursor, page_offset, length);
1174
		break;
1175
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1176
		page = ceph_msg_data_pages_next(cursor, page_offset, length);
1177
		break;
1178 1179
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1180
		page = ceph_msg_data_bio_next(cursor, page_offset, length);
1181
		break;
1182
#endif /* CONFIG_BLOCK */
1183 1184 1185
	case CEPH_MSG_DATA_BVECS:
		page = ceph_msg_data_bvecs_next(cursor, page_offset, length);
		break;
1186
	case CEPH_MSG_DATA_NONE:
1187 1188 1189 1190
	default:
		page = NULL;
		break;
	}
1191

1192 1193 1194
	BUG_ON(!page);
	BUG_ON(*page_offset + *length > PAGE_SIZE);
	BUG_ON(!*length);
1195
	BUG_ON(*length > cursor->resid);
1196
	if (last_piece)
A
Alex Elder 已提交
1197
		*last_piece = cursor->last_piece;
1198 1199 1200 1201 1202 1203 1204 1205

	return page;
}

/*
 * Returns true if the result moves the cursor on to the next piece
 * of the data item.
 */
1206 1207
static void ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor,
				  size_t bytes)
1208 1209 1210
{
	bool new_piece;

1211
	BUG_ON(bytes > cursor->resid);
A
Alex Elder 已提交
1212
	switch (cursor->data->type) {
1213
	case CEPH_MSG_DATA_PAGELIST:
A
Alex Elder 已提交
1214
		new_piece = ceph_msg_data_pagelist_advance(cursor, bytes);
1215
		break;
1216
	case CEPH_MSG_DATA_PAGES:
A
Alex Elder 已提交
1217
		new_piece = ceph_msg_data_pages_advance(cursor, bytes);
1218
		break;
1219 1220
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
A
Alex Elder 已提交
1221
		new_piece = ceph_msg_data_bio_advance(cursor, bytes);
1222
		break;
1223
#endif /* CONFIG_BLOCK */
1224 1225 1226
	case CEPH_MSG_DATA_BVECS:
		new_piece = ceph_msg_data_bvecs_advance(cursor, bytes);
		break;
1227
	case CEPH_MSG_DATA_NONE:
1228 1229 1230 1231
	default:
		BUG();
		break;
	}
1232
	cursor->total_resid -= bytes;
1233

1234 1235 1236
	if (!cursor->resid && cursor->total_resid) {
		WARN_ON(!cursor->last_piece);
		BUG_ON(list_is_last(&cursor->data->links, cursor->data_head));
1237
		cursor->data = list_next_entry(cursor->data, links);
1238
		__ceph_msg_data_cursor_init(cursor);
A
Alex Elder 已提交
1239
		new_piece = true;
1240
	}
A
Alex Elder 已提交
1241
	cursor->need_crc = new_piece;
1242 1243
}

1244 1245 1246 1247 1248 1249 1250
static size_t sizeof_footer(struct ceph_connection *con)
{
	return (con->peer_features & CEPH_FEATURE_MSG_AUTH) ?
	    sizeof(struct ceph_msg_footer) :
	    sizeof(struct ceph_msg_footer_old);
}

1251
static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
1252 1253
{
	BUG_ON(!msg);
1254
	BUG_ON(!data_len);
1255

1256
	/* Initialize data cursor */
1257

A
Alex Elder 已提交
1258
	ceph_msg_data_cursor_init(msg, (size_t)data_len);
1259 1260
}

S
Sage Weil 已提交
1261 1262 1263 1264
/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
1265
static void prepare_write_message_footer(struct ceph_connection *con)
S
Sage Weil 已提交
1266 1267 1268
{
	struct ceph_msg *m = con->out_msg;

1269 1270
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

S
Sage Weil 已提交
1271
	dout("prepare_write_message_footer %p\n", con);
I
Ilya Dryomov 已提交
1272
	con_out_kvec_add(con, sizeof_footer(con), &m->footer);
Y
Yan, Zheng 已提交
1273 1274
	if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
		if (con->ops->sign_message)
1275
			con->ops->sign_message(m);
Y
Yan, Zheng 已提交
1276 1277 1278 1279 1280
		else
			m->footer.sig = 0;
	} else {
		m->old_footer.flags = m->footer.flags;
	}
S
Sage Weil 已提交
1281
	con->out_more = m->more_to_follow;
1282
	con->out_msg_done = true;
S
Sage Weil 已提交
1283 1284 1285 1286 1287 1288 1289 1290
}

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

1293
	con_out_kvec_reset(con);
1294
	con->out_msg_done = false;
S
Sage Weil 已提交
1295 1296 1297 1298 1299

	/* 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;
1300
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
S
Sage Weil 已提交
1301
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1302
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1303
			&con->out_temp_ack);
S
Sage Weil 已提交
1304 1305
	}

1306
	BUG_ON(list_empty(&con->out_queue));
1307
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
1308
	con->out_msg = m;
1309
	BUG_ON(m->con != con);
1310 1311 1312 1313

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

1315 1316 1317 1318 1319 1320 1321
	/*
	 * 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;
1322

1323 1324 1325
		if (con->ops->reencode_message)
			con->ops->reencode_message(m);
	}
S
Sage Weil 已提交
1326

1327
	dout("prepare_write_message %p seq %lld type %d len %d+%d+%zd\n",
S
Sage Weil 已提交
1328 1329
	     m, con->out_seq, le16_to_cpu(m->hdr.type),
	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
1330
	     m->data_length);
1331 1332
	WARN_ON(m->front.iov_len != le32_to_cpu(m->hdr.front_len));
	WARN_ON(m->data_length != le32_to_cpu(m->hdr.data_len));
S
Sage Weil 已提交
1333 1334

	/* tag + hdr + front + middle */
1335
	con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
I
Ilya Dryomov 已提交
1336
	con_out_kvec_add(con, sizeof(con->out_hdr), &con->out_hdr);
1337
	con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
1338

S
Sage Weil 已提交
1339
	if (m->middle)
1340
		con_out_kvec_add(con, m->middle->vec.iov_len,
1341
			m->middle->vec.iov_base);
S
Sage Weil 已提交
1342

I
Ilya Dryomov 已提交
1343
	/* fill in hdr crc and finalize hdr */
1344 1345
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
I
Ilya Dryomov 已提交
1346
	memcpy(&con->out_hdr, &con->out_msg->hdr, sizeof(con->out_hdr));
1347

I
Ilya Dryomov 已提交
1348
	/* fill in front and middle crc, footer */
1349 1350 1351 1352 1353 1354 1355
	crc = crc32c(0, m->front.iov_base, m->front.iov_len);
	con->out_msg->footer.front_crc = cpu_to_le32(crc);
	if (m->middle) {
		crc = crc32c(0, m->middle->vec.iov_base,
				m->middle->vec.iov_len);
		con->out_msg->footer.middle_crc = cpu_to_le32(crc);
	} else
S
Sage Weil 已提交
1356
		con->out_msg->footer.middle_crc = 0;
1357
	dout("%s front_crc %u middle_crc %u\n", __func__,
S
Sage Weil 已提交
1358 1359
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));
I
Ilya Dryomov 已提交
1360
	con->out_msg->footer.flags = 0;
S
Sage Weil 已提交
1361 1362

	/* is there a data payload? */
1363
	con->out_msg->footer.data_crc = 0;
1364 1365
	if (m->data_length) {
		prepare_message_data(con->out_msg, m->data_length);
1366 1367
		con->out_more = 1;  /* data + footer will follow */
	} else {
S
Sage Weil 已提交
1368
		/* no, queue up footer too and be done */
1369
		prepare_write_message_footer(con);
1370
	}
S
Sage Weil 已提交
1371

1372
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383
}

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

1384
	con_out_kvec_reset(con);
1385

1386
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1387

S
Sage Weil 已提交
1388
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1389
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1390 1391
				&con->out_temp_ack);

S
Sage Weil 已提交
1392
	con->out_more = 1;  /* more will follow.. eventually.. */
1393
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1394 1395
}

1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413
/*
 * Prepare to share the seq during handshake
 */
static void prepare_write_seq(struct ceph_connection *con)
{
	dout("prepare_write_seq %p %llu -> %llu\n", con,
	     con->in_seq_acked, con->in_seq);
	con->in_seq_acked = con->in_seq;

	con_out_kvec_reset(con);

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

	con_flag_set(con, CON_FLAG_WRITE_PENDING);
}

S
Sage Weil 已提交
1414 1415 1416 1417 1418 1419
/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
1420
	con_out_kvec_reset(con);
1421
	if (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2) {
1422
		struct timespec now;
1423

1424
		ktime_get_real_ts(&now);
1425
		con_out_kvec_add(con, sizeof(tag_keepalive2), &tag_keepalive2);
1426 1427 1428
		ceph_encode_timespec(&con->out_temp_keepalive2, &now);
		con_out_kvec_add(con, sizeof(con->out_temp_keepalive2),
				 &con->out_temp_keepalive2);
1429 1430 1431
	} else {
		con_out_kvec_add(con, sizeof(tag_keepalive), &tag_keepalive);
	}
1432
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
1433 1434 1435 1436 1437 1438
}

/*
 * Connection negotiation.
 */

1439 1440
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
1441
{
1442
	struct ceph_auth_handshake *auth;
1443 1444 1445 1446

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;
1447
		return NULL;
1448 1449
	}

1450
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1451
	if (IS_ERR(auth))
1452
		return auth;
1453

1454 1455
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1456
	return auth;
1457 1458
}

S
Sage Weil 已提交
1459 1460 1461
/*
 * We connected to a peer and are saying hello.
 */
1462
static void prepare_write_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1463
{
1464 1465
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1466
					&con->msgr->my_enc_addr);
1467 1468

	con->out_more = 0;
1469
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1470 1471
}

1472
static int prepare_write_connect(struct ceph_connection *con)
1473
{
1474
	unsigned int global_seq = get_global_seq(con->msgr, 0);
S
Sage Weil 已提交
1475
	int proto;
1476
	int auth_proto;
1477
	struct ceph_auth_handshake *auth;
S
Sage Weil 已提交
1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494

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

1496 1497
	con->out_connect.features =
	    cpu_to_le64(from_msgr(con->msgr)->supported_features);
S
Sage Weil 已提交
1498 1499 1500 1501 1502 1503
	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;

1504 1505
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
1506 1507
	if (IS_ERR(auth))
		return PTR_ERR(auth);
1508

1509
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1510 1511 1512
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

1513
	con_out_kvec_add(con, sizeof (con->out_connect),
1514 1515
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
1516
		con_out_kvec_add(con, auth->authorizer_buf_len,
1517
					auth->authorizer_buf);
1518

S
Sage Weil 已提交
1519
	con->out_more = 0;
1520
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1521

1522
	return 0;
S
Sage Weil 已提交
1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544
}

/*
 * 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 */
1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556

		/* account for full iov entries consumed */
		while (ret >= con->out_kvec_cur->iov_len) {
			BUG_ON(!con->out_kvec_left);
			ret -= con->out_kvec_cur->iov_len;
			con->out_kvec_cur++;
			con->out_kvec_left--;
		}
		/* and for a partially-consumed entry */
		if (ret) {
			con->out_kvec_cur->iov_len -= ret;
			con->out_kvec_cur->iov_base += ret;
S
Sage Weil 已提交
1557 1558 1559 1560 1561 1562 1563 1564 1565 1566
		}
	}
	con->out_kvec_left = 0;
	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! */
}

1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579
static u32 ceph_crc32c_page(u32 crc, struct page *page,
				unsigned int page_offset,
				unsigned int length)
{
	char *kaddr;

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

	return crc;
}
S
Sage Weil 已提交
1580 1581 1582 1583 1584 1585 1586
/*
 * 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
 */
1587
static int write_partial_message_data(struct ceph_connection *con)
S
Sage Weil 已提交
1588 1589
{
	struct ceph_msg *msg = con->out_msg;
A
Alex Elder 已提交
1590
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1591
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
A
Alex Elder 已提交
1592
	u32 crc;
S
Sage Weil 已提交
1593

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

1596
	if (list_empty(&msg->data))
1597 1598
		return -EINVAL;

1599 1600 1601 1602 1603 1604 1605 1606
	/*
	 * Iterate through each page that contains data to be
	 * written, and send as much as possible for each.
	 *
	 * If we are calculating the data crc (the default), we will
	 * need to map the page.  If we have no pages, they have
	 * been revoked, so use the zero page.
	 */
A
Alex Elder 已提交
1607
	crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0;
1608
	while (cursor->resid) {
1609
		struct page *page;
1610 1611
		size_t page_offset;
		size_t length;
1612
		bool last_piece;
A
Alex Elder 已提交
1613
		int ret;
1614

1615 1616
		page = ceph_msg_data_next(cursor, &page_offset, &length,
					  &last_piece);
1617
		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1618
					length, !last_piece);
A
Alex Elder 已提交
1619 1620 1621
		if (ret <= 0) {
			if (do_datacrc)
				msg->footer.data_crc = cpu_to_le32(crc);
S
Sage Weil 已提交
1622

A
Alex Elder 已提交
1623 1624
			return ret;
		}
1625 1626
		if (do_datacrc && cursor->need_crc)
			crc = ceph_crc32c_page(crc, page, page_offset, length);
1627
		ceph_msg_data_advance(cursor, (size_t)ret);
S
Sage Weil 已提交
1628 1629
	}

1630
	dout("%s %p msg %p done\n", __func__, con, msg);
S
Sage Weil 已提交
1631 1632

	/* prepare and queue up footer, too */
A
Alex Elder 已提交
1633 1634 1635
	if (do_datacrc)
		msg->footer.data_crc = cpu_to_le32(crc);
	else
1636
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1637
	con_out_kvec_reset(con);
1638
	prepare_write_message_footer(con);
A
Alex Elder 已提交
1639 1640

	return 1;	/* must return > 0 to indicate success */
S
Sage Weil 已提交
1641 1642 1643 1644 1645 1646 1647 1648 1649
}

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

I
Ilya Dryomov 已提交
1650
	dout("%s %p %d left\n", __func__, con, con->out_skip);
S
Sage Weil 已提交
1651
	while (con->out_skip > 0) {
1652
		size_t size = min(con->out_skip, (int) PAGE_SIZE);
S
Sage Weil 已提交
1653

1654
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
S
Sage Weil 已提交
1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1667 1668 1669 1670 1671 1672
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684
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;
}

1685 1686 1687 1688 1689 1690 1691
static void prepare_read_seq(struct ceph_connection *con)
{
	dout("prepare_read_seq %p\n", con);
	con->in_base_pos = 0;
	con->in_tag = CEPH_MSGR_TAG_SEQ;
}

S
Sage Weil 已提交
1692 1693 1694 1695 1696 1697 1698
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;
}

1699 1700 1701 1702 1703 1704
static void prepare_read_keepalive_ack(struct ceph_connection *con)
{
	dout("prepare_read_keepalive_ack %p\n", con);
	con->in_base_pos = 0;
}

S
Sage Weil 已提交
1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718
/*
 * 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,
1719
			int end, int size, void *object)
S
Sage Weil 已提交
1720
{
1721 1722
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
S
Sage Weil 已提交
1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735
		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
 */
1736
static int read_partial_banner(struct ceph_connection *con)
S
Sage Weil 已提交
1737
{
1738 1739 1740
	int size;
	int end;
	int ret;
S
Sage Weil 已提交
1741

1742
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
S
Sage Weil 已提交
1743 1744

	/* peer's banner */
1745 1746 1747
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
S
Sage Weil 已提交
1748 1749
	if (ret <= 0)
		goto out;
1750 1751 1752 1753

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
S
Sage Weil 已提交
1754 1755
	if (ret <= 0)
		goto out;
1756 1757 1758 1759

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

1763 1764 1765 1766 1767 1768
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1769 1770 1771
	int size;
	int end;
	int ret;
1772 1773 1774

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

1775 1776 1777
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
S
Sage Weil 已提交
1778 1779
	if (ret <= 0)
		goto out;
1780 1781 1782 1783

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

1787 1788 1789
	dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
	     con, (int)con->in_reply.tag,
	     le32_to_cpu(con->in_reply.connect_seq),
S
Sage Weil 已提交
1790 1791 1792
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1793

S
Sage Weil 已提交
1794 1795 1796 1797 1798 1799 1800 1801
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1802
		pr_err("connect to %s got bad banner\n",
1803
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
1804 1805 1806 1807 1808 1809 1810 1811
		con->error_msg = "protocol error, bad banner";
		return -1;
	}
	return 0;
}

static bool addr_is_blank(struct sockaddr_storage *ss)
{
1812 1813 1814
	struct in_addr *addr = &((struct sockaddr_in *)ss)->sin_addr;
	struct in6_addr *addr6 = &((struct sockaddr_in6 *)ss)->sin6_addr;

S
Sage Weil 已提交
1815 1816
	switch (ss->ss_family) {
	case AF_INET:
1817
		return addr->s_addr == htonl(INADDR_ANY);
S
Sage Weil 已提交
1818
	case AF_INET6:
1819 1820 1821
		return ipv6_addr_any(addr6);
	default:
		return true;
S
Sage Weil 已提交
1822 1823 1824 1825 1826 1827 1828
	}
}

static int addr_port(struct sockaddr_storage *ss)
{
	switch (ss->ss_family) {
	case AF_INET:
S
Sage Weil 已提交
1829
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
S
Sage Weil 已提交
1830
	case AF_INET6:
S
Sage Weil 已提交
1831
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
S
Sage Weil 已提交
1832 1833 1834 1835 1836 1837 1838 1839 1840
	}
	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);
1841
		break;
S
Sage Weil 已提交
1842 1843
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1844
		break;
S
Sage Weil 已提交
1845 1846 1847
	}
}

N
Noah Watkins 已提交
1848 1849 1850 1851 1852 1853
/*
 * Unlike other *_pton function semantics, zero indicates success.
 */
static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
		char delim, const char **ipend)
{
A
Alex Elder 已提交
1854 1855
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
N
Noah Watkins 已提交
1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942

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

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

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

	return -EINVAL;
}

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

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

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

	if (end <= name)
		return -EINVAL;

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

	kfree(ip_addr);

	*ipend = end;

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

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

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

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

	return ret;
}

S
Sage Weil 已提交
1943 1944 1945 1946 1947 1948 1949 1950
/*
 * Parse an ip[:port] list into an addr array.  Use the default
 * monitor port if a port isn't specified.
 */
int ceph_parse_ips(const char *c, const char *end,
		   struct ceph_entity_addr *addr,
		   int max_count, int *count)
{
N
Noah Watkins 已提交
1951
	int i, ret = -EINVAL;
S
Sage Weil 已提交
1952 1953 1954 1955 1956 1957 1958
	const char *p = c;

	dout("parse_ips on '%.*s'\n", (int)(end-c), c);
	for (i = 0; i < max_count; i++) {
		const char *ipend;
		struct sockaddr_storage *ss = &addr[i].in_addr;
		int port;
S
Sage Weil 已提交
1959 1960 1961 1962 1963 1964
		char delim = ',';

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

N
Noah Watkins 已提交
1966 1967
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
S
Sage Weil 已提交
1968
			goto bad;
N
Noah Watkins 已提交
1969 1970
		ret = -EINVAL;

S
Sage Weil 已提交
1971 1972
		p = ipend;

S
Sage Weil 已提交
1973 1974 1975 1976 1977 1978 1979 1980
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

S
Sage Weil 已提交
1981 1982 1983 1984 1985 1986 1987 1988
		/* port? */
		if (p < end && *p == ':') {
			port = 0;
			p++;
			while (p < end && *p >= '0' && *p <= '9') {
				port = (port * 10) + (*p - '0');
				p++;
			}
1989 1990 1991
			if (port == 0)
				port = CEPH_MON_PORT;
			else if (port > 65535)
S
Sage Weil 已提交
1992 1993 1994 1995 1996 1997 1998
				goto bad;
		} else {
			port = CEPH_MON_PORT;
		}

		addr_set_port(ss, port);

1999
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
S
Sage Weil 已提交
2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015

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

	if (p != end)
		goto bad;

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

bad:
S
Sage Weil 已提交
2016
	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
N
Noah Watkins 已提交
2017
	return ret;
S
Sage Weil 已提交
2018
}
2019
EXPORT_SYMBOL(ceph_parse_ips);
S
Sage Weil 已提交
2020

2021
static int process_banner(struct ceph_connection *con)
S
Sage Weil 已提交
2022
{
2023
	dout("process_banner on %p\n", con);
S
Sage Weil 已提交
2024 2025 2026 2027

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

2028 2029 2030
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

S
Sage Weil 已提交
2031 2032 2033 2034 2035
	/*
	 * Make sure the other end is who we wanted.  note that the other
	 * end may not yet know their ip address, so if it's 0.0.0.0, give
	 * them the benefit of the doubt.
	 */
S
Sage Weil 已提交
2036 2037
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
S
Sage Weil 已提交
2038 2039
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
2040 2041 2042 2043 2044
		pr_warn("wrong peer, want %s/%d, got %s/%d\n",
			ceph_pr_addr(&con->peer_addr.in_addr),
			(int)le32_to_cpu(con->peer_addr.nonce),
			ceph_pr_addr(&con->actual_peer_addr.in_addr),
			(int)le32_to_cpu(con->actual_peer_addr.nonce));
2045
		con->error_msg = "wrong peer at address";
S
Sage Weil 已提交
2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058
		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);
2059
		encode_my_addr(con->msgr);
2060
		dout("process_banner learned my addr is %s\n",
2061
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
S
Sage Weil 已提交
2062 2063
	}

2064 2065 2066 2067 2068
	return 0;
}

static int process_connect(struct ceph_connection *con)
{
2069 2070
	u64 sup_feat = from_msgr(con->msgr)->supported_features;
	u64 req_feat = from_msgr(con->msgr)->required_features;
2071
	u64 server_feat = le64_to_cpu(con->in_reply.features);
2072
	int ret;
2073

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

2076 2077 2078 2079 2080 2081
	if (con->auth_reply_buf) {
		/*
		 * Any connection that defines ->get_authorizer()
		 * should also define ->verify_authorizer_reply().
		 * See get_connect_authorizer().
		 */
2082
		ret = con->ops->verify_authorizer_reply(con);
2083 2084 2085 2086 2087 2088
		if (ret < 0) {
			con->error_msg = "bad authorize reply";
			return ret;
		}
	}

S
Sage Weil 已提交
2089
	switch (con->in_reply.tag) {
2090 2091 2092 2093
	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),
2094
		       ceph_pr_addr(&con->peer_addr.in_addr),
2095 2096
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
2097
		reset_connection(con);
2098 2099
		return -1;

S
Sage Weil 已提交
2100 2101 2102 2103
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
2104
		       ceph_pr_addr(&con->peer_addr.in_addr),
S
Sage Weil 已提交
2105 2106 2107
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
2108
		reset_connection(con);
S
Sage Weil 已提交
2109 2110
		return -1;

2111 2112 2113 2114 2115 2116 2117 2118
	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;
		}
2119
		con_out_kvec_reset(con);
2120
		ret = prepare_write_connect(con);
2121 2122
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2123
		prepare_read_connect(con);
2124
		break;
S
Sage Weil 已提交
2125 2126 2127 2128 2129 2130 2131 2132 2133 2134

	case CEPH_MSGR_TAG_RESETSESSION:
		/*
		 * If we connected with a large connect_seq but the peer
		 * has no record of a session with us (no connection, or
		 * connect_seq == 0), they will send RESETSESION to indicate
		 * that they must have reset their session, and may have
		 * dropped messages.
		 */
		dout("process_connect got RESET peer seq %u\n",
S
Sage Weil 已提交
2135
		     le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2136 2137
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
2138
		       ceph_pr_addr(&con->peer_addr.in_addr));
S
Sage Weil 已提交
2139
		reset_connection(con);
2140
		con_out_kvec_reset(con);
2141 2142 2143
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2144 2145 2146
		prepare_read_connect(con);

		/* Tell ceph about it. */
2147
		mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2148 2149 2150
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
2151
		mutex_lock(&con->mutex);
2152
		if (con->state != CON_STATE_NEGOTIATING)
2153
			return -EAGAIN;
S
Sage Weil 已提交
2154 2155 2156 2157 2158 2159 2160
		break;

	case CEPH_MSGR_TAG_RETRY_SESSION:
		/*
		 * If we sent a smaller connect_seq than the peer has, try
		 * again with a larger value.
		 */
S
Sage Weil 已提交
2161
		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
S
Sage Weil 已提交
2162
		     le32_to_cpu(con->out_connect.connect_seq),
S
Sage Weil 已提交
2163 2164
		     le32_to_cpu(con->in_reply.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2165
		con_out_kvec_reset(con);
2166 2167 2168
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2169 2170 2171 2172 2173 2174 2175 2176
		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.
		 */
2177
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
S
Sage Weil 已提交
2178
		     con->peer_global_seq,
S
Sage Weil 已提交
2179
		     le32_to_cpu(con->in_reply.global_seq));
S
Sage Weil 已提交
2180
		get_global_seq(con->msgr,
S
Sage Weil 已提交
2181
			       le32_to_cpu(con->in_reply.global_seq));
2182
		con_out_kvec_reset(con);
2183 2184 2185
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
S
Sage Weil 已提交
2186 2187 2188
		prepare_read_connect(con);
		break;

2189
	case CEPH_MSGR_TAG_SEQ:
S
Sage Weil 已提交
2190
	case CEPH_MSGR_TAG_READY:
2191 2192 2193 2194
		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),
2195
			       ceph_pr_addr(&con->peer_addr.in_addr),
2196 2197
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
2198
			reset_connection(con);
2199 2200
			return -1;
		}
2201

2202
		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2203
		con->state = CON_STATE_OPEN;
2204
		con->auth_retry = 0;    /* we authenticated; clear flag */
S
Sage Weil 已提交
2205 2206
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
2207
		con->peer_features = server_feat;
S
Sage Weil 已提交
2208 2209 2210 2211 2212 2213
		dout("process_connect got READY gseq %d cseq %d (%d)\n",
		     con->peer_global_seq,
		     le32_to_cpu(con->in_reply.connect_seq),
		     con->connect_seq);
		WARN_ON(con->connect_seq !=
			le32_to_cpu(con->in_reply.connect_seq));
S
Sage Weil 已提交
2214 2215

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

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

2220 2221 2222 2223 2224 2225
		if (con->in_reply.tag == CEPH_MSGR_TAG_SEQ) {
			prepare_write_seq(con);
			prepare_read_seq(con);
		} else {
			prepare_read_tag(con);
		}
S
Sage Weil 已提交
2226 2227 2228 2229 2230 2231 2232 2233 2234
		break;

	case CEPH_MSGR_TAG_WAIT:
		/*
		 * If there is a connection race (we are opening
		 * connections to each other), one of us may just have
		 * to WAIT.  This shouldn't happen if we are the
		 * client.
		 */
S
Sage Weil 已提交
2235 2236
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
S
Sage Weil 已提交
2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250

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


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

2254
	return read_partial(con, end, size, &con->in_temp_ack);
S
Sage Weil 已提交
2255 2256 2257 2258 2259 2260 2261 2262 2263 2264
}

/*
 * 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;
2265 2266
	bool reconnect = (con->in_tag == CEPH_MSGR_TAG_SEQ);
	struct list_head *list = reconnect ? &con->out_queue : &con->out_sent;
S
Sage Weil 已提交
2267

2268 2269 2270 2271 2272 2273 2274 2275 2276
	/*
	 * In the reconnect case, con_fault() has requeued messages
	 * in out_sent. We should cleanup old messages according to
	 * the reconnect seq.
	 */
	while (!list_empty(list)) {
		m = list_first_entry(list, struct ceph_msg, list_head);
		if (reconnect && m->needs_out_seq)
			break;
S
Sage Weil 已提交
2277 2278 2279 2280 2281
		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);
2282
		m->ack_stamp = jiffies;
S
Sage Weil 已提交
2283 2284
		ceph_msg_remove(m);
	}
2285

S
Sage Weil 已提交
2286 2287 2288 2289
	prepare_read_tag(con);
}


2290
static int read_partial_message_section(struct ceph_connection *con,
S
Sage Weil 已提交
2291 2292
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
2293
{
2294
	int ret, left;
2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306

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

2310 2311
	return 1;
}
S
Sage Weil 已提交
2312

2313 2314 2315
static int read_partial_msg_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->in_msg;
A
Alex Elder 已提交
2316
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
2317
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
A
Alex Elder 已提交
2318 2319 2320
	struct page *page;
	size_t page_offset;
	size_t length;
A
Alex Elder 已提交
2321
	u32 crc = 0;
2322 2323 2324
	int ret;

	BUG_ON(!msg);
2325
	if (list_empty(&msg->data))
2326
		return -EIO;
2327

A
Alex Elder 已提交
2328 2329
	if (do_datacrc)
		crc = con->in_data_crc;
2330
	while (cursor->resid) {
2331
		page = ceph_msg_data_next(cursor, &page_offset, &length, NULL);
A
Alex Elder 已提交
2332
		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
A
Alex Elder 已提交
2333 2334 2335 2336
		if (ret <= 0) {
			if (do_datacrc)
				con->in_data_crc = crc;

A
Alex Elder 已提交
2337
			return ret;
A
Alex Elder 已提交
2338
		}
A
Alex Elder 已提交
2339 2340

		if (do_datacrc)
A
Alex Elder 已提交
2341
			crc = ceph_crc32c_page(crc, page, page_offset, ret);
2342
		ceph_msg_data_advance(cursor, (size_t)ret);
2343
	}
A
Alex Elder 已提交
2344 2345
	if (do_datacrc)
		con->in_data_crc = crc;
2346 2347 2348 2349

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

S
Sage Weil 已提交
2350 2351 2352
/*
 * read (part of) a message.
 */
A
Alex Elder 已提交
2353 2354
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);

S
Sage Weil 已提交
2355 2356 2357
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
2358 2359
	int size;
	int end;
S
Sage Weil 已提交
2360
	int ret;
2361
	unsigned int front_len, middle_len, data_len;
2362
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
Y
Yan, Zheng 已提交
2363
	bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
2364
	u64 seq;
2365
	u32 crc;
S
Sage Weil 已提交
2366 2367 2368 2369

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

	/* header */
2370 2371 2372
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
2373 2374
	if (ret <= 0)
		return ret;
2375 2376 2377

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

S
Sage Weil 已提交
2383 2384 2385 2386
	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);
2387
	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
S
Sage Weil 已提交
2388 2389 2390 2391 2392
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

2393 2394 2395
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
S
Sage Weil 已提交
2396
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2397
			ENTITY_NAME(con->peer_name),
2398
			ceph_pr_addr(&con->peer_addr.in_addr),
2399 2400
			seq, con->in_seq + 1);
		con->in_base_pos = -front_len - middle_len - data_len -
2401
			sizeof_footer(con);
2402
		con->in_tag = CEPH_MSGR_TAG_READY;
2403
		return 1;
2404 2405 2406 2407
	} 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";
2408
		return -EBADE;
2409 2410
	}

S
Sage Weil 已提交
2411 2412
	/* allocate message? */
	if (!con->in_msg) {
2413 2414
		int skip = 0;

S
Sage Weil 已提交
2415
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2416
		     front_len, data_len);
2417 2418 2419
		ret = ceph_con_in_msg_alloc(con, &skip);
		if (ret < 0)
			return ret;
2420 2421

		BUG_ON(!con->in_msg ^ skip);
2422
		if (skip) {
S
Sage Weil 已提交
2423
			/* skip this message */
2424
			dout("alloc_msg said skip message\n");
S
Sage Weil 已提交
2425
			con->in_base_pos = -front_len - middle_len - data_len -
2426
				sizeof_footer(con);
S
Sage Weil 已提交
2427
			con->in_tag = CEPH_MSGR_TAG_READY;
2428
			con->in_seq++;
2429
			return 1;
S
Sage Weil 已提交
2430
		}
2431

2432
		BUG_ON(!con->in_msg);
2433
		BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
2434 2435
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
2436 2437
		if (m->middle)
			m->middle->vec.iov_len = 0;
2438

2439
		/* prepare for data payload, if any */
2440

2441
		if (data_len)
2442
			prepare_message_data(con->in_msg, data_len);
S
Sage Weil 已提交
2443 2444 2445
	}

	/* front */
2446 2447 2448 2449
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
S
Sage Weil 已提交
2450 2451

	/* middle */
2452
	if (m->middle) {
S
Sage Weil 已提交
2453 2454
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
2455
						   &con->in_middle_crc);
S
Sage Weil 已提交
2456 2457 2458 2459 2460
		if (ret <= 0)
			return ret;
	}

	/* (page) data */
2461 2462 2463 2464
	if (data_len) {
		ret = read_partial_msg_data(con);
		if (ret <= 0)
			return ret;
S
Sage Weil 已提交
2465 2466 2467
	}

	/* footer */
I
Ilya Dryomov 已提交
2468
	size = sizeof_footer(con);
2469 2470
	end += size;
	ret = read_partial(con, end, size, &m->footer);
2471 2472 2473
	if (ret <= 0)
		return ret;

Y
Yan, Zheng 已提交
2474 2475 2476 2477 2478
	if (!need_sign) {
		m->footer.flags = m->old_footer.flags;
		m->footer.sig = 0;
	}

S
Sage Weil 已提交
2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493
	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;
	}
2494
	if (do_datacrc &&
S
Sage Weil 已提交
2495 2496 2497 2498 2499 2500 2501
	    (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
	    con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
		pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
		       con->in_data_crc, le32_to_cpu(m->footer.data_crc));
		return -EBADMSG;
	}

Y
Yan, Zheng 已提交
2502
	if (need_sign && con->ops->check_message_signature &&
2503
	    con->ops->check_message_signature(m)) {
Y
Yan, Zheng 已提交
2504 2505 2506 2507
		pr_err("read_partial_message %p signature check failed\n", m);
		return -EBADMSG;
	}

S
Sage Weil 已提交
2508 2509 2510 2511 2512 2513 2514 2515 2516 2517
	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)
{
2518
	struct ceph_msg *msg = con->in_msg;
S
Sage Weil 已提交
2519

2520
	BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
2521 2522 2523 2524
	con->in_msg = NULL;

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

	con->in_seq++;
2528
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
2529 2530 2531

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
2532
	     ENTITY_NAME(msg->hdr.src),
S
Sage Weil 已提交
2533 2534 2535 2536 2537 2538
	     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);
2539 2540

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

2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553
static int read_keepalive_ack(struct ceph_connection *con)
{
	struct ceph_timespec ceph_ts;
	size_t size = sizeof(ceph_ts);
	int ret = read_partial(con, size, size, &ceph_ts);
	if (ret <= 0)
		return ret;
	ceph_decode_timespec(&con->last_keepalive_ack, &ceph_ts);
	prepare_read_tag(con);
	return 1;
}
S
Sage Weil 已提交
2554 2555 2556 2557 2558 2559 2560 2561 2562

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

2563
	dout("try_write start %p state %lu\n", con, con->state);
S
Sage Weil 已提交
2564 2565 2566 2567 2568

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

	/* open the socket first? */
2569 2570 2571
	if (con->state == CON_STATE_PREOPEN) {
		BUG_ON(con->sock);
		con->state = CON_STATE_CONNECTING;
2572

2573
		con_out_kvec_reset(con);
2574
		prepare_write_banner(con);
2575
		prepare_read_banner(con);
S
Sage Weil 已提交
2576

2577
		BUG_ON(con->in_msg);
S
Sage Weil 已提交
2578 2579 2580
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
2581 2582
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
S
Sage Weil 已提交
2583 2584 2585 2586 2587 2588 2589
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
I
Ilya Dryomov 已提交
2590 2591
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
S
Sage Weil 已提交
2592
		if (ret <= 0)
2593
			goto out;
S
Sage Weil 已提交
2594
	}
I
Ilya Dryomov 已提交
2595 2596
	if (con->out_skip) {
		ret = write_partial_skip(con);
S
Sage Weil 已提交
2597
		if (ret <= 0)
2598
			goto out;
S
Sage Weil 已提交
2599 2600 2601 2602
	}

	/* msg pages? */
	if (con->out_msg) {
2603 2604 2605 2606 2607 2608
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

2609
		ret = write_partial_message_data(con);
S
Sage Weil 已提交
2610 2611 2612
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2613
			goto out;
S
Sage Weil 已提交
2614
		if (ret < 0) {
2615
			dout("try_write write_partial_message_data err %d\n",
S
Sage Weil 已提交
2616
			     ret);
2617
			goto out;
S
Sage Weil 已提交
2618 2619 2620
		}
	}

2621
do_next:
2622
	if (con->state == CON_STATE_OPEN) {
2623 2624 2625 2626
		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
			prepare_write_keepalive(con);
			goto more;
		}
S
Sage Weil 已提交
2627 2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638
		/* is anything else pending? */
		if (!list_empty(&con->out_queue)) {
			prepare_write_message(con);
			goto more;
		}
		if (con->in_seq > con->in_seq_acked) {
			prepare_write_ack(con);
			goto more;
		}
	}

	/* Nothing to do! */
2639
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
S
Sage Weil 已提交
2640 2641 2642
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2643
	dout("try_write done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655
	return ret;
}



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

2656 2657 2658 2659 2660
more:
	dout("try_read start on %p state %lu\n", con, con->state);
	if (con->state != CON_STATE_CONNECTING &&
	    con->state != CON_STATE_NEGOTIATING &&
	    con->state != CON_STATE_OPEN)
S
Sage Weil 已提交
2661 2662
		return 0;

2663
	BUG_ON(!con->sock);
2664

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

2668
	if (con->state == CON_STATE_CONNECTING) {
2669 2670 2671
		dout("try_read connecting\n");
		ret = read_partial_banner(con);
		if (ret <= 0)
2672
			goto out;
2673 2674 2675 2676
		ret = process_banner(con);
		if (ret < 0)
			goto out;

2677
		con->state = CON_STATE_NEGOTIATING;
2678

2679 2680 2681 2682 2683
		/*
		 * Received banner is good, exchange connection info.
		 * Do not reset out_kvec, as sending our banner raced
		 * with receiving peer banner after connect completed.
		 */
2684 2685 2686 2687 2688 2689
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
		prepare_read_connect(con);

		/* Send connection info before awaiting response */
2690 2691 2692
		goto out;
	}

2693
	if (con->state == CON_STATE_NEGOTIATING) {
2694
		dout("try_read negotiating\n");
S
Sage Weil 已提交
2695 2696 2697
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2698 2699 2700
		ret = process_connect(con);
		if (ret < 0)
			goto out;
S
Sage Weil 已提交
2701 2702 2703
		goto more;
	}

2704
	WARN_ON(con->state != CON_STATE_OPEN);
2705

S
Sage Weil 已提交
2706 2707 2708 2709 2710 2711
	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
A
Alex Elder 已提交
2712 2713 2714
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

S
Sage Weil 已提交
2715 2716 2717
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2718
			goto out;
S
Sage Weil 已提交
2719 2720 2721 2722 2723 2724 2725 2726 2727 2728
		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)
2729
			goto out;
S
Sage Weil 已提交
2730 2731 2732 2733 2734 2735 2736 2737
		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;
2738 2739 2740
		case CEPH_MSGR_TAG_KEEPALIVE2_ACK:
			prepare_read_keepalive_ack(con);
			break;
S
Sage Weil 已提交
2741
		case CEPH_MSGR_TAG_CLOSE:
2742 2743
			con_close_socket(con);
			con->state = CON_STATE_CLOSED;
2744
			goto out;
S
Sage Weil 已提交
2745 2746 2747 2748 2749 2750 2751 2752 2753
		default:
			goto bad_tag;
		}
	}
	if (con->in_tag == CEPH_MSGR_TAG_MSG) {
		ret = read_partial_message(con);
		if (ret <= 0) {
			switch (ret) {
			case -EBADMSG:
2754
				con->error_msg = "bad crc/signature";
2755 2756
				/* fall through */
			case -EBADE:
S
Sage Weil 已提交
2757
				ret = -EIO;
2758
				break;
S
Sage Weil 已提交
2759 2760
			case -EIO:
				con->error_msg = "io error";
2761
				break;
S
Sage Weil 已提交
2762
			}
2763
			goto out;
S
Sage Weil 已提交
2764 2765 2766 2767
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
2768 2769
		if (con->state == CON_STATE_OPEN)
			prepare_read_tag(con);
S
Sage Weil 已提交
2770 2771
		goto more;
	}
2772 2773 2774 2775 2776 2777
	if (con->in_tag == CEPH_MSGR_TAG_ACK ||
	    con->in_tag == CEPH_MSGR_TAG_SEQ) {
		/*
		 * the final handshake seq exchange is semantically
		 * equivalent to an ACK
		 */
S
Sage Weil 已提交
2778 2779
		ret = read_partial_ack(con);
		if (ret <= 0)
2780
			goto out;
S
Sage Weil 已提交
2781 2782 2783
		process_ack(con);
		goto more;
	}
2784 2785 2786 2787 2788 2789
	if (con->in_tag == CEPH_MSGR_TAG_KEEPALIVE2_ACK) {
		ret = read_keepalive_ack(con);
		if (ret <= 0)
			goto out;
		goto more;
	}
S
Sage Weil 已提交
2790 2791

out:
2792
	dout("try_read done on %p ret %d\n", con, ret);
S
Sage Weil 已提交
2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803
	return ret;

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


/*
A
Alex Elder 已提交
2804 2805 2806
 * Atomically queue work on a connection after the specified delay.
 * Bump @con reference to avoid races with connection teardown.
 * Returns 0 if work was queued, or an error code otherwise.
S
Sage Weil 已提交
2807
 */
A
Alex Elder 已提交
2808
static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
S
Sage Weil 已提交
2809 2810
{
	if (!con->ops->get(con)) {
A
Alex Elder 已提交
2811 2812
		dout("%s %p ref count 0\n", __func__, con);
		return -ENOENT;
S
Sage Weil 已提交
2813 2814
	}

A
Alex Elder 已提交
2815 2816
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
		dout("%s %p - already queued\n", __func__, con);
S
Sage Weil 已提交
2817
		con->ops->put(con);
A
Alex Elder 已提交
2818
		return -EBUSY;
S
Sage Weil 已提交
2819
	}
A
Alex Elder 已提交
2820 2821 2822 2823 2824 2825 2826 2827

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

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

2830 2831 2832 2833 2834 2835 2836 2837
static void cancel_con(struct ceph_connection *con)
{
	if (cancel_delayed_work(&con->work)) {
		dout("%s %p\n", __func__, con);
		con->ops->put(con);
	}
}

2838 2839
static bool con_sock_closed(struct ceph_connection *con)
{
2840
	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855
		return false;

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

	switch (con->state) {
	CASE(CLOSED);
	CASE(PREOPEN);
	CASE(CONNECTING);
	CASE(NEGOTIATING);
	CASE(OPEN);
	CASE(STANDBY);
	default:
2856
		pr_warn("%s con %p unrecognized state %lu\n",
2857 2858 2859 2860 2861 2862 2863 2864 2865 2866
			__func__, con, con->state);
		con->error_msg = "unrecognized con state";
		BUG();
		break;
	}
#undef CASE

	return true;
}

2867 2868 2869 2870 2871 2872 2873 2874 2875 2876 2877 2878 2879 2880 2881 2882 2883 2884
static bool con_backoff(struct ceph_connection *con)
{
	int ret;

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

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

	return true;
}

2885 2886 2887 2888
/* Finish fault handling; con->mutex must *not* be held here */

static void con_fault_finish(struct ceph_connection *con)
{
2889 2890
	dout("%s %p\n", __func__, con);

2891 2892 2893 2894
	/*
	 * in case we faulted due to authentication, invalidate our
	 * current tickets so that we can get new ones.
	 */
2895 2896 2897 2898 2899
	if (con->auth_retry) {
		dout("auth_retry %d, invalidating\n", con->auth_retry);
		if (con->ops->invalidate_authorizer)
			con->ops->invalidate_authorizer(con);
		con->auth_retry = 0;
2900 2901 2902 2903 2904 2905
	}

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

S
Sage Weil 已提交
2906 2907 2908
/*
 * Do some work on a connection.  Drop a connection ref when we're done.
 */
2909
static void ceph_con_workfn(struct work_struct *work)
S
Sage Weil 已提交
2910 2911 2912
{
	struct ceph_connection *con = container_of(work, struct ceph_connection,
						   work.work);
2913
	bool fault;
S
Sage Weil 已提交
2914

S
Sage Weil 已提交
2915
	mutex_lock(&con->mutex);
2916 2917
	while (true) {
		int ret;
S
Sage Weil 已提交
2918

2919 2920 2921 2922 2923 2924 2925 2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939
		if ((fault = con_sock_closed(con))) {
			dout("%s: con %p SOCK_CLOSED\n", __func__, con);
			break;
		}
		if (con_backoff(con)) {
			dout("%s: con %p BACKOFF\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_STANDBY) {
			dout("%s: con %p STANDBY\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_CLOSED) {
			dout("%s: con %p CLOSED\n", __func__, con);
			BUG_ON(con->sock);
			break;
		}
		if (con->state == CON_STATE_PREOPEN) {
			dout("%s: con %p PREOPEN\n", __func__, con);
			BUG_ON(con->sock);
		}
2940

2941 2942 2943 2944
		ret = try_read(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2945 2946
			if (!con->error_msg)
				con->error_msg = "socket error on read";
2947 2948 2949 2950 2951 2952 2953 2954
			fault = true;
			break;
		}

		ret = try_write(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2955 2956
			if (!con->error_msg)
				con->error_msg = "socket error on write";
2957 2958 2959 2960
			fault = true;
		}

		break;	/* If we make it to here, we're done */
2961
	}
2962 2963
	if (fault)
		con_fault(con);
S
Sage Weil 已提交
2964
	mutex_unlock(&con->mutex);
2965

2966 2967 2968 2969
	if (fault)
		con_fault_finish(con);

	con->ops->put(con);
S
Sage Weil 已提交
2970 2971 2972 2973 2974 2975
}

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

2981 2982 2983 2984
	pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
		ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
	con->error_msg = NULL;

2985
	WARN_ON(con->state != CON_STATE_CONNECTING &&
2986 2987
	       con->state != CON_STATE_NEGOTIATING &&
	       con->state != CON_STATE_OPEN);
2988

S
Sage Weil 已提交
2989
	con_close_socket(con);
S
Sage Weil 已提交
2990

2991
	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2992 2993
		dout("fault on LOSSYTX channel, marking CLOSED\n");
		con->state = CON_STATE_CLOSED;
2994
		return;
2995 2996
	}

S
Sage Weil 已提交
2997
	if (con->in_msg) {
2998
		BUG_ON(con->in_msg->con != con);
S
Sage Weil 已提交
2999 3000 3001
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
	}
S
Sage Weil 已提交
3002

3003 3004
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
3005

S
Sage Weil 已提交
3006 3007 3008
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
3009
	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
S
Sage Weil 已提交
3010
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
3011
		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
3012
		con->state = CON_STATE_STANDBY;
3013 3014
	} else {
		/* retry after a delay. */
3015
		con->state = CON_STATE_PREOPEN;
3016 3017 3018 3019
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
3020
		con_flag_set(con, CON_FLAG_BACKOFF);
A
Alex Elder 已提交
3021
		queue_con(con);
S
Sage Weil 已提交
3022 3023 3024 3025 3026 3027
	}
}



/*
3028
 * initialize a new messenger instance
S
Sage Weil 已提交
3029
 */
3030
void ceph_messenger_init(struct ceph_messenger *msgr,
3031
			 struct ceph_entity_addr *myaddr)
S
Sage Weil 已提交
3032 3033 3034 3035 3036 3037 3038
{
	spin_lock_init(&msgr->global_seq_lock);

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

	/* select a random nonce */
3039
	msgr->inst.addr.type = 0;
S
Sage Weil 已提交
3040
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
3041
	encode_my_addr(msgr);
S
Sage Weil 已提交
3042

3043
	atomic_set(&msgr->stopping, 0);
3044
	write_pnet(&msgr->net, get_net(current->nsproxy->net_ns));
S
Sage Weil 已提交
3045

3046
	dout("%s %p\n", __func__, msgr);
S
Sage Weil 已提交
3047
}
3048
EXPORT_SYMBOL(ceph_messenger_init);
S
Sage Weil 已提交
3049

3050 3051 3052 3053 3054 3055
void ceph_messenger_fini(struct ceph_messenger *msgr)
{
	put_net(read_pnet(&msgr->net));
}
EXPORT_SYMBOL(ceph_messenger_fini);

3056 3057 3058 3059 3060 3061 3062 3063 3064
static void msg_con_set(struct ceph_msg *msg, struct ceph_connection *con)
{
	if (msg->con)
		msg->con->ops->put(msg->con);

	msg->con = con ? con->ops->get(con) : NULL;
	BUG_ON(msg->con != con);
}

S
Sage Weil 已提交
3065 3066 3067
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
3068
	if (con->state == CON_STATE_STANDBY) {
S
Sage Weil 已提交
3069
		dout("clear_standby %p and ++connect_seq\n", con);
3070
		con->state = CON_STATE_PREOPEN;
S
Sage Weil 已提交
3071
		con->connect_seq++;
3072 3073
		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
S
Sage Weil 已提交
3074 3075 3076
	}
}

S
Sage Weil 已提交
3077 3078 3079 3080 3081 3082
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	/* set src+dst */
3083
	msg->hdr.src = con->msgr->inst.name;
3084
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
3085 3086
	msg->needs_out_seq = true;

3087
	mutex_lock(&con->mutex);
3088

3089
	if (con->state == CON_STATE_CLOSED) {
3090 3091 3092 3093 3094 3095
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		mutex_unlock(&con->mutex);
		return;
	}

3096
	msg_con_set(msg, con);
3097

S
Sage Weil 已提交
3098 3099 3100 3101 3102 3103 3104 3105
	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));
3106 3107

	clear_standby(con);
3108
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3109 3110 3111

	/* if there wasn't anything waiting to send before, queue
	 * new work */
3112
	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
3113 3114
		queue_con(con);
}
3115
EXPORT_SYMBOL(ceph_con_send);
S
Sage Weil 已提交
3116 3117 3118 3119

/*
 * Revoke a message that was previously queued for send
 */
3120
void ceph_msg_revoke(struct ceph_msg *msg)
S
Sage Weil 已提交
3121
{
3122 3123
	struct ceph_connection *con = msg->con;

3124 3125
	if (!con) {
		dout("%s msg %p null con\n", __func__, msg);
3126
		return;		/* Message not in our possession */
3127
	}
3128

3129
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3130
	if (!list_empty(&msg->list_head)) {
3131
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
S
Sage Weil 已提交
3132 3133
		list_del_init(&msg->list_head);
		msg->hdr.seq = 0;
3134

S
Sage Weil 已提交
3135
		ceph_msg_put(msg);
S
Sage Weil 已提交
3136 3137
	}
	if (con->out_msg == msg) {
I
Ilya Dryomov 已提交
3138 3139 3140 3141 3142 3143
		BUG_ON(con->out_skip);
		/* footer */
		if (con->out_msg_done) {
			con->out_skip += con_out_kvec_skip(con);
		} else {
			BUG_ON(!msg->data_length);
I
Ilya Dryomov 已提交
3144
			con->out_skip += sizeof_footer(con);
S
Sage Weil 已提交
3145
		}
I
Ilya Dryomov 已提交
3146 3147 3148 3149 3150 3151 3152 3153 3154
		/* data, middle, front */
		if (msg->data_length)
			con->out_skip += msg->cursor.total_resid;
		if (msg->middle)
			con->out_skip += con_out_kvec_skip(con);
		con->out_skip += con_out_kvec_skip(con);

		dout("%s %p msg %p - was sending, will write %d skip %d\n",
		     __func__, con, msg, con->out_kvec_bytes, con->out_skip);
S
Sage Weil 已提交
3155
		msg->hdr.seq = 0;
I
Ilya Dryomov 已提交
3156
		con->out_msg = NULL;
3157
		ceph_msg_put(msg);
S
Sage Weil 已提交
3158
	}
I
Ilya Dryomov 已提交
3159

3160
	mutex_unlock(&con->mutex);
S
Sage Weil 已提交
3161 3162
}

3163
/*
3164
 * Revoke a message that we may be reading data into
3165
 */
3166
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
3167
{
3168
	struct ceph_connection *con = msg->con;
3169

3170
	if (!con) {
3171 3172 3173 3174
		dout("%s msg %p null con\n", __func__, msg);
		return;		/* Message not in our possession */
	}

3175
	mutex_lock(&con->mutex);
3176
	if (con->in_msg == msg) {
3177 3178 3179
		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);
3180 3181

		/* skip rest of message */
3182 3183
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
3184
				sizeof(struct ceph_msg_header) -
3185 3186 3187
				front_len -
				middle_len -
				data_len -
3188 3189 3190 3191
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
3192
		con->in_seq++;
3193
	} else {
3194 3195
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
3196 3197 3198 3199
	}
	mutex_unlock(&con->mutex);
}

S
Sage Weil 已提交
3200 3201 3202 3203 3204
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
S
Sage Weil 已提交
3205
	dout("con_keepalive %p\n", con);
3206
	mutex_lock(&con->mutex);
S
Sage Weil 已提交
3207
	clear_standby(con);
3208
	mutex_unlock(&con->mutex);
3209 3210
	if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 &&
	    con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
S
Sage Weil 已提交
3211 3212
		queue_con(con);
}
3213
EXPORT_SYMBOL(ceph_con_keepalive);
S
Sage Weil 已提交
3214

3215 3216 3217 3218 3219
bool ceph_con_keepalive_expired(struct ceph_connection *con,
			       unsigned long interval)
{
	if (interval > 0 &&
	    (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2)) {
3220
		struct timespec now;
3221
		struct timespec ts;
3222
		ktime_get_real_ts(&now);
3223 3224 3225 3226 3227 3228 3229
		jiffies_to_timespec(interval, &ts);
		ts = timespec_add(con->last_keepalive_ack, ts);
		return timespec_compare(&now, &ts) >= 0;
	}
	return false;
}

3230
static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type)
A
Alex Elder 已提交
3231
{
3232 3233 3234 3235 3236
	struct ceph_msg_data *data;

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

3237
	data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS);
3238 3239 3240 3241
	if (!data)
		return NULL;

	data->type = type;
3242
	INIT_LIST_HEAD(&data->links);
3243 3244 3245 3246 3247 3248 3249 3250 3251

	return data;
}

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

3252
	WARN_ON(!list_empty(&data->links));
Y
Yan, Zheng 已提交
3253
	if (data->type == CEPH_MSG_DATA_PAGELIST)
3254
		ceph_pagelist_release(data->pagelist);
3255
	kmem_cache_free(ceph_msg_data_cache, data);
A
Alex Elder 已提交
3256 3257
}

3258
void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
3259
		size_t length, size_t alignment)
3260
{
3261 3262
	struct ceph_msg_data *data;

3263 3264
	BUG_ON(!pages);
	BUG_ON(!length);
3265 3266 3267 3268 3269 3270

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

3272 3273
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3274
}
3275
EXPORT_SYMBOL(ceph_msg_data_add_pages);
S
Sage Weil 已提交
3276

3277
void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
3278 3279
				struct ceph_pagelist *pagelist)
{
3280 3281
	struct ceph_msg_data *data;

3282 3283
	BUG_ON(!pagelist);
	BUG_ON(!pagelist->length);
3284

3285 3286 3287 3288
	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
	BUG_ON(!data);
	data->pagelist = pagelist;

3289 3290
	list_add_tail(&data->links, &msg->data);
	msg->data_length += pagelist->length;
3291
}
3292
EXPORT_SYMBOL(ceph_msg_data_add_pagelist);
3293

3294
#ifdef	CONFIG_BLOCK
3295 3296
void ceph_msg_data_add_bio(struct ceph_msg *msg, struct ceph_bio_iter *bio_pos,
			   u32 length)
3297
{
3298 3299 3300 3301
	struct ceph_msg_data *data;

	data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
	BUG_ON(!data);
3302
	data->bio_pos = *bio_pos;
A
Alex Elder 已提交
3303
	data->bio_length = length;
3304

3305 3306
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3307
}
3308
EXPORT_SYMBOL(ceph_msg_data_add_bio);
3309
#endif	/* CONFIG_BLOCK */
3310

3311 3312 3313 3314 3315 3316 3317 3318 3319 3320 3321 3322 3323 3324
void ceph_msg_data_add_bvecs(struct ceph_msg *msg,
			     struct ceph_bvec_iter *bvec_pos)
{
	struct ceph_msg_data *data;

	data = ceph_msg_data_create(CEPH_MSG_DATA_BVECS);
	BUG_ON(!data);
	data->bvec_pos = *bvec_pos;

	list_add_tail(&data->links, &msg->data);
	msg->data_length += bvec_pos->iter.bi_size;
}
EXPORT_SYMBOL(ceph_msg_data_add_bvecs);

S
Sage Weil 已提交
3325 3326 3327 3328
/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
3329 3330
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
S
Sage Weil 已提交
3331 3332 3333
{
	struct ceph_msg *m;

3334
	m = kmem_cache_zalloc(ceph_msg_cache, flags);
S
Sage Weil 已提交
3335 3336 3337 3338
	if (m == NULL)
		goto out;

	m->hdr.type = cpu_to_le16(type);
3339
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
S
Sage Weil 已提交
3340
	m->hdr.front_len = cpu_to_le32(front_len);
3341

3342 3343
	INIT_LIST_HEAD(&m->list_head);
	kref_init(&m->kref);
3344
	INIT_LIST_HEAD(&m->data);
3345

S
Sage Weil 已提交
3346 3347
	/* front */
	if (front_len) {
3348
		m->front.iov_base = ceph_kvmalloc(front_len, flags);
S
Sage Weil 已提交
3349
		if (m->front.iov_base == NULL) {
3350
			dout("ceph_msg_new can't allocate %d bytes\n",
S
Sage Weil 已提交
3351 3352 3353 3354 3355 3356
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
3357
	m->front_alloc_len = m->front.iov_len = front_len;
S
Sage Weil 已提交
3358

S
Sage Weil 已提交
3359
	dout("ceph_msg_new %p front %d\n", m, front_len);
S
Sage Weil 已提交
3360 3361 3362 3363 3364
	return m;

out2:
	ceph_msg_put(m);
out:
3365 3366 3367
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
3368
		WARN_ON(1);
3369 3370 3371 3372
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
3373
	return NULL;
S
Sage Weil 已提交
3374
}
3375
EXPORT_SYMBOL(ceph_msg_new);
S
Sage Weil 已提交
3376 3377 3378 3379 3380 3381 3382 3383

/*
 * 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.
 */
3384
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
S
Sage Weil 已提交
3385 3386 3387 3388 3389 3390 3391 3392 3393
{
	int type = le16_to_cpu(msg->hdr.type);
	int middle_len = le32_to_cpu(msg->hdr.middle_len);

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

S
Sage Weil 已提交
3394
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
S
Sage Weil 已提交
3395 3396 3397 3398 3399
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

3400
/*
A
Alex Elder 已提交
3401 3402 3403 3404
 * 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.
 *
3405 3406 3407 3408 3409 3410 3411 3412 3413
 * Returns 0 on success, or a negative error code.
 *
 * On success, if we set *skip = 1:
 *  - the next message should be skipped and ignored.
 *  - con->in_msg == NULL
 * or if we set *skip = 0:
 *  - con->in_msg is non-null.
 * On error (ENOMEM, EAGAIN, ...),
 *  - con->in_msg == NULL
3414
 */
3415
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3416
{
3417
	struct ceph_msg_header *hdr = &con->in_hdr;
3418
	int middle_len = le32_to_cpu(hdr->middle_len);
3419
	struct ceph_msg *msg;
3420
	int ret = 0;
3421

A
Alex Elder 已提交
3422
	BUG_ON(con->in_msg != NULL);
3423
	BUG_ON(!con->ops->alloc_msg);
3424

3425 3426 3427 3428 3429
	mutex_unlock(&con->mutex);
	msg = con->ops->alloc_msg(con, hdr, skip);
	mutex_lock(&con->mutex);
	if (con->state != CON_STATE_OPEN) {
		if (msg)
3430
			ceph_msg_put(msg);
3431 3432
		return -EAGAIN;
	}
3433 3434
	if (msg) {
		BUG_ON(*skip);
3435
		msg_con_set(msg, con);
3436 3437 3438 3439 3440 3441 3442 3443 3444 3445
		con->in_msg = msg;
	} else {
		/*
		 * Null message pointer means either we should skip
		 * this message or we couldn't allocate memory.  The
		 * former is not an error.
		 */
		if (*skip)
			return 0;

3446
		con->error_msg = "error allocating memory for incoming message";
3447
		return -ENOMEM;
3448
	}
A
Alex Elder 已提交
3449
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3450

A
Alex Elder 已提交
3451 3452
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
3453
		if (ret < 0) {
A
Alex Elder 已提交
3454 3455
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
3456 3457
		}
	}
3458

3459
	return ret;
3460 3461
}

S
Sage Weil 已提交
3462 3463 3464 3465

/*
 * Free a generically kmalloc'd message.
 */
3466
static void ceph_msg_free(struct ceph_msg *m)
S
Sage Weil 已提交
3467
{
3468
	dout("%s %p\n", __func__, m);
I
Ilya Dryomov 已提交
3469
	kvfree(m->front.iov_base);
3470
	kmem_cache_free(ceph_msg_cache, m);
S
Sage Weil 已提交
3471 3472
}

3473
static void ceph_msg_release(struct kref *kref)
S
Sage Weil 已提交
3474 3475
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
3476
	struct ceph_msg_data *data, *next;
S
Sage Weil 已提交
3477

3478
	dout("%s %p\n", __func__, m);
S
Sage Weil 已提交
3479 3480
	WARN_ON(!list_empty(&m->list_head));

3481 3482
	msg_con_set(m, NULL);

S
Sage Weil 已提交
3483 3484 3485 3486
	/* drop middle, data, if any */
	if (m->middle) {
		ceph_buffer_put(m->middle);
		m->middle = NULL;
S
Sage Weil 已提交
3487
	}
3488

3489 3490
	list_for_each_entry_safe(data, next, &m->data, links) {
		list_del_init(&data->links);
3491 3492
		ceph_msg_data_destroy(data);
	}
A
Alex Elder 已提交
3493
	m->data_length = 0;
3494

S
Sage Weil 已提交
3495 3496 3497
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
3498 3499 3500 3501 3502 3503
		ceph_msg_free(m);
}

struct ceph_msg *ceph_msg_get(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
3504
	     kref_read(&msg->kref));
3505 3506 3507 3508 3509 3510 3511 3512
	kref_get(&msg->kref);
	return msg;
}
EXPORT_SYMBOL(ceph_msg_get);

void ceph_msg_put(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
3513
	     kref_read(&msg->kref));
3514
	kref_put(&msg->kref, ceph_msg_release);
S
Sage Weil 已提交
3515
}
3516
EXPORT_SYMBOL(ceph_msg_put);
3517 3518 3519

void ceph_msg_dump(struct ceph_msg *msg)
{
3520 3521
	pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg,
		 msg->front_alloc_len, msg->data_length);
3522 3523 3524 3525 3526 3527 3528 3529 3530 3531 3532 3533 3534 3535 3536
	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);
}
3537
EXPORT_SYMBOL(ceph_msg_dump);