libceph: small refactor in write_partial_kvec()
[linux-2.6.git] / net / ceph / messenger.c
1 #include <linux/ceph/ceph_debug.h>
2
3 #include <linux/crc32c.h>
4 #include <linux/ctype.h>
5 #include <linux/highmem.h>
6 #include <linux/inet.h>
7 #include <linux/kthread.h>
8 #include <linux/net.h>
9 #include <linux/slab.h>
10 #include <linux/socket.h>
11 #include <linux/string.h>
12 #include <linux/bio.h>
13 #include <linux/blkdev.h>
14 #include <linux/dns_resolver.h>
15 #include <net/tcp.h>
16
17 #include <linux/ceph/libceph.h>
18 #include <linux/ceph/messenger.h>
19 #include <linux/ceph/decode.h>
20 #include <linux/ceph/pagelist.h>
21 #include <linux/export.h>
22
23 /*
24  * Ceph uses the messenger to exchange ceph_msg messages with other
25  * hosts in the system.  The messenger provides ordered and reliable
26  * delivery.  We tolerate TCP disconnects by reconnecting (with
27  * exponential backoff) in the case of a fault (disconnection, bad
28  * crc, protocol error).  Acks allow sent messages to be discarded by
29  * the sender.
30  */
31
32 /* static tag bytes (protocol control messages) */
33 static char tag_msg = CEPH_MSGR_TAG_MSG;
34 static char tag_ack = CEPH_MSGR_TAG_ACK;
35 static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
36
37 #ifdef CONFIG_LOCKDEP
38 static struct lock_class_key socket_class;
39 #endif
40
41
42 static void queue_con(struct ceph_connection *con);
43 static void con_work(struct work_struct *);
44 static void ceph_fault(struct ceph_connection *con);
45
46 /*
47  * Nicely render a sockaddr as a string.  An array of formatted
48  * strings is used, to approximate reentrancy.
49  */
50 #define ADDR_STR_COUNT_LOG      5       /* log2(# address strings in array) */
51 #define ADDR_STR_COUNT          (1 << ADDR_STR_COUNT_LOG)
52 #define ADDR_STR_COUNT_MASK     (ADDR_STR_COUNT - 1)
53 #define MAX_ADDR_STR_LEN        64      /* 54 is enough */
54
55 static char addr_str[ADDR_STR_COUNT][MAX_ADDR_STR_LEN];
56 static atomic_t addr_str_seq = ATOMIC_INIT(0);
57
58 static struct page *zero_page;          /* used in certain error cases */
59 static void *zero_page_address;         /* kernel virtual addr of zero_page */
60
61 const char *ceph_pr_addr(const struct sockaddr_storage *ss)
62 {
63         int i;
64         char *s;
65         struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
66         struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
67
68         i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
69         s = addr_str[i];
70
71         switch (ss->ss_family) {
72         case AF_INET:
73                 snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
74                          ntohs(in4->sin_port));
75                 break;
76
77         case AF_INET6:
78                 snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
79                          ntohs(in6->sin6_port));
80                 break;
81
82         default:
83                 snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
84                          ss->ss_family);
85         }
86
87         return s;
88 }
89 EXPORT_SYMBOL(ceph_pr_addr);
90
91 static void encode_my_addr(struct ceph_messenger *msgr)
92 {
93         memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
94         ceph_encode_addr(&msgr->my_enc_addr);
95 }
96
97 /*
98  * work queue for all reading and writing to/from the socket.
99  */
100 static struct workqueue_struct *ceph_msgr_wq;
101
102 void _ceph_msgr_exit(void)
103 {
104         if (ceph_msgr_wq) {
105                 destroy_workqueue(ceph_msgr_wq);
106                 ceph_msgr_wq = NULL;
107         }
108
109         BUG_ON(zero_page_address == NULL);
110         zero_page_address = NULL;
111
112         BUG_ON(zero_page == NULL);
113         kunmap(zero_page);
114         page_cache_release(zero_page);
115         zero_page = NULL;
116 }
117
118 int ceph_msgr_init(void)
119 {
120         BUG_ON(zero_page != NULL);
121         zero_page = ZERO_PAGE(0);
122         page_cache_get(zero_page);
123
124         BUG_ON(zero_page_address != NULL);
125         zero_page_address = kmap(zero_page);
126
127         ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_NON_REENTRANT, 0);
128         if (ceph_msgr_wq)
129                 return 0;
130
131         pr_err("msgr_init failed to create workqueue\n");
132         _ceph_msgr_exit();
133
134         return -ENOMEM;
135 }
136 EXPORT_SYMBOL(ceph_msgr_init);
137
138 void ceph_msgr_exit(void)
139 {
140         BUG_ON(ceph_msgr_wq == NULL);
141
142         _ceph_msgr_exit();
143 }
144 EXPORT_SYMBOL(ceph_msgr_exit);
145
146 void ceph_msgr_flush(void)
147 {
148         flush_workqueue(ceph_msgr_wq);
149 }
150 EXPORT_SYMBOL(ceph_msgr_flush);
151
152
153 /*
154  * socket callback functions
155  */
156
157 /* data available on socket, or listen socket received a connect */
158 static void ceph_data_ready(struct sock *sk, int count_unused)
159 {
160         struct ceph_connection *con = sk->sk_user_data;
161
162         if (sk->sk_state != TCP_CLOSE_WAIT) {
163                 dout("ceph_data_ready on %p state = %lu, queueing work\n",
164                      con, con->state);
165                 queue_con(con);
166         }
167 }
168
169 /* socket has buffer space for writing */
170 static void ceph_write_space(struct sock *sk)
171 {
172         struct ceph_connection *con = sk->sk_user_data;
173
174         /* only queue to workqueue if there is data we want to write,
175          * and there is sufficient space in the socket buffer to accept
176          * more data.  clear SOCK_NOSPACE so that ceph_write_space()
177          * doesn't get called again until try_write() fills the socket
178          * buffer. See net/ipv4/tcp_input.c:tcp_check_space()
179          * and net/core/stream.c:sk_stream_write_space().
180          */
181         if (test_bit(WRITE_PENDING, &con->state)) {
182                 if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
183                         dout("ceph_write_space %p queueing write work\n", con);
184                         clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
185                         queue_con(con);
186                 }
187         } else {
188                 dout("ceph_write_space %p nothing to write\n", con);
189         }
190 }
191
192 /* socket's state has changed */
193 static void ceph_state_change(struct sock *sk)
194 {
195         struct ceph_connection *con = sk->sk_user_data;
196
197         dout("ceph_state_change %p state = %lu sk_state = %u\n",
198              con, con->state, sk->sk_state);
199
200         if (test_bit(CLOSED, &con->state))
201                 return;
202
203         switch (sk->sk_state) {
204         case TCP_CLOSE:
205                 dout("ceph_state_change TCP_CLOSE\n");
206         case TCP_CLOSE_WAIT:
207                 dout("ceph_state_change TCP_CLOSE_WAIT\n");
208                 if (test_and_set_bit(SOCK_CLOSED, &con->state) == 0) {
209                         if (test_bit(CONNECTING, &con->state))
210                                 con->error_msg = "connection failed";
211                         else
212                                 con->error_msg = "socket closed";
213                         queue_con(con);
214                 }
215                 break;
216         case TCP_ESTABLISHED:
217                 dout("ceph_state_change TCP_ESTABLISHED\n");
218                 queue_con(con);
219                 break;
220         default:        /* Everything else is uninteresting */
221                 break;
222         }
223 }
224
225 /*
226  * set up socket callbacks
227  */
228 static void set_sock_callbacks(struct socket *sock,
229                                struct ceph_connection *con)
230 {
231         struct sock *sk = sock->sk;
232         sk->sk_user_data = con;
233         sk->sk_data_ready = ceph_data_ready;
234         sk->sk_write_space = ceph_write_space;
235         sk->sk_state_change = ceph_state_change;
236 }
237
238
239 /*
240  * socket helpers
241  */
242
243 /*
244  * initiate connection to a remote socket.
245  */
246 static int ceph_tcp_connect(struct ceph_connection *con)
247 {
248         struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
249         struct socket *sock;
250         int ret;
251
252         BUG_ON(con->sock);
253         ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
254                                IPPROTO_TCP, &sock);
255         if (ret)
256                 return ret;
257         sock->sk->sk_allocation = GFP_NOFS;
258
259 #ifdef CONFIG_LOCKDEP
260         lockdep_set_class(&sock->sk->sk_lock, &socket_class);
261 #endif
262
263         set_sock_callbacks(sock, con);
264
265         dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
266
267         ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
268                                  O_NONBLOCK);
269         if (ret == -EINPROGRESS) {
270                 dout("connect %s EINPROGRESS sk_state = %u\n",
271                      ceph_pr_addr(&con->peer_addr.in_addr),
272                      sock->sk->sk_state);
273         } else if (ret < 0) {
274                 pr_err("connect %s error %d\n",
275                        ceph_pr_addr(&con->peer_addr.in_addr), ret);
276                 sock_release(sock);
277                 con->error_msg = "connect error";
278
279                 return ret;
280         }
281         con->sock = sock;
282
283         return 0;
284 }
285
286 static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
287 {
288         struct kvec iov = {buf, len};
289         struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
290         int r;
291
292         r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
293         if (r == -EAGAIN)
294                 r = 0;
295         return r;
296 }
297
298 /*
299  * write something.  @more is true if caller will be sending more data
300  * shortly.
301  */
302 static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
303                      size_t kvlen, size_t len, int more)
304 {
305         struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
306         int r;
307
308         if (more)
309                 msg.msg_flags |= MSG_MORE;
310         else
311                 msg.msg_flags |= MSG_EOR;  /* superfluous, but what the hell */
312
313         r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
314         if (r == -EAGAIN)
315                 r = 0;
316         return r;
317 }
318
319
320 /*
321  * Shutdown/close the socket for the given connection.
322  */
323 static int con_close_socket(struct ceph_connection *con)
324 {
325         int rc;
326
327         dout("con_close_socket on %p sock %p\n", con, con->sock);
328         if (!con->sock)
329                 return 0;
330         set_bit(SOCK_CLOSED, &con->state);
331         rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
332         sock_release(con->sock);
333         con->sock = NULL;
334         clear_bit(SOCK_CLOSED, &con->state);
335         return rc;
336 }
337
338 /*
339  * Reset a connection.  Discard all incoming and outgoing messages
340  * and clear *_seq state.
341  */
342 static void ceph_msg_remove(struct ceph_msg *msg)
343 {
344         list_del_init(&msg->list_head);
345         ceph_msg_put(msg);
346 }
347 static void ceph_msg_remove_list(struct list_head *head)
348 {
349         while (!list_empty(head)) {
350                 struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
351                                                         list_head);
352                 ceph_msg_remove(msg);
353         }
354 }
355
356 static void reset_connection(struct ceph_connection *con)
357 {
358         /* reset connection, out_queue, msg_ and connect_seq */
359         /* discard existing out_queue and msg_seq */
360         ceph_msg_remove_list(&con->out_queue);
361         ceph_msg_remove_list(&con->out_sent);
362
363         if (con->in_msg) {
364                 ceph_msg_put(con->in_msg);
365                 con->in_msg = NULL;
366         }
367
368         con->connect_seq = 0;
369         con->out_seq = 0;
370         if (con->out_msg) {
371                 ceph_msg_put(con->out_msg);
372                 con->out_msg = NULL;
373         }
374         con->in_seq = 0;
375         con->in_seq_acked = 0;
376 }
377
378 /*
379  * mark a peer down.  drop any open connections.
380  */
381 void ceph_con_close(struct ceph_connection *con)
382 {
383         dout("con_close %p peer %s\n", con,
384              ceph_pr_addr(&con->peer_addr.in_addr));
385         set_bit(CLOSED, &con->state);  /* in case there's queued work */
386         clear_bit(STANDBY, &con->state);  /* avoid connect_seq bump */
387         clear_bit(LOSSYTX, &con->state);  /* so we retry next connect */
388         clear_bit(KEEPALIVE_PENDING, &con->state);
389         clear_bit(WRITE_PENDING, &con->state);
390         mutex_lock(&con->mutex);
391         reset_connection(con);
392         con->peer_global_seq = 0;
393         cancel_delayed_work(&con->work);
394         mutex_unlock(&con->mutex);
395         queue_con(con);
396 }
397 EXPORT_SYMBOL(ceph_con_close);
398
399 /*
400  * Reopen a closed connection, with a new peer address.
401  */
402 void ceph_con_open(struct ceph_connection *con, struct ceph_entity_addr *addr)
403 {
404         dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
405         set_bit(OPENING, &con->state);
406         clear_bit(CLOSED, &con->state);
407         memcpy(&con->peer_addr, addr, sizeof(*addr));
408         con->delay = 0;      /* reset backoff memory */
409         queue_con(con);
410 }
411 EXPORT_SYMBOL(ceph_con_open);
412
413 /*
414  * return true if this connection ever successfully opened
415  */
416 bool ceph_con_opened(struct ceph_connection *con)
417 {
418         return con->connect_seq > 0;
419 }
420
421 /*
422  * generic get/put
423  */
424 struct ceph_connection *ceph_con_get(struct ceph_connection *con)
425 {
426         int nref = __atomic_add_unless(&con->nref, 1, 0);
427
428         dout("con_get %p nref = %d -> %d\n", con, nref, nref + 1);
429
430         return nref ? con : NULL;
431 }
432
433 void ceph_con_put(struct ceph_connection *con)
434 {
435         int nref = atomic_dec_return(&con->nref);
436
437         BUG_ON(nref < 0);
438         if (nref == 0) {
439                 BUG_ON(con->sock);
440                 kfree(con);
441         }
442         dout("con_put %p nref = %d -> %d\n", con, nref + 1, nref);
443 }
444
445 /*
446  * initialize a new connection.
447  */
448 void ceph_con_init(struct ceph_messenger *msgr, struct ceph_connection *con)
449 {
450         dout("con_init %p\n", con);
451         memset(con, 0, sizeof(*con));
452         atomic_set(&con->nref, 1);
453         con->msgr = msgr;
454         mutex_init(&con->mutex);
455         INIT_LIST_HEAD(&con->out_queue);
456         INIT_LIST_HEAD(&con->out_sent);
457         INIT_DELAYED_WORK(&con->work, con_work);
458 }
459 EXPORT_SYMBOL(ceph_con_init);
460
461
462 /*
463  * We maintain a global counter to order connection attempts.  Get
464  * a unique seq greater than @gt.
465  */
466 static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
467 {
468         u32 ret;
469
470         spin_lock(&msgr->global_seq_lock);
471         if (msgr->global_seq < gt)
472                 msgr->global_seq = gt;
473         ret = ++msgr->global_seq;
474         spin_unlock(&msgr->global_seq_lock);
475         return ret;
476 }
477
478 static void ceph_con_out_kvec_reset(struct ceph_connection *con)
479 {
480         con->out_kvec_left = 0;
481         con->out_kvec_bytes = 0;
482         con->out_kvec_cur = &con->out_kvec[0];
483 }
484
485 static void ceph_con_out_kvec_add(struct ceph_connection *con,
486                                 size_t size, void *data)
487 {
488         int index;
489
490         index = con->out_kvec_left;
491         BUG_ON(index >= ARRAY_SIZE(con->out_kvec));
492
493         con->out_kvec[index].iov_len = size;
494         con->out_kvec[index].iov_base = data;
495         con->out_kvec_left++;
496         con->out_kvec_bytes += size;
497 }
498
499 /*
500  * Prepare footer for currently outgoing message, and finish things
501  * off.  Assumes out_kvec* are already valid.. we just add on to the end.
502  */
503 static void prepare_write_message_footer(struct ceph_connection *con)
504 {
505         struct ceph_msg *m = con->out_msg;
506         int v = con->out_kvec_left;
507
508         dout("prepare_write_message_footer %p\n", con);
509         con->out_kvec_is_msg = true;
510         con->out_kvec[v].iov_base = &m->footer;
511         con->out_kvec[v].iov_len = sizeof(m->footer);
512         con->out_kvec_bytes += sizeof(m->footer);
513         con->out_kvec_left++;
514         con->out_more = m->more_to_follow;
515         con->out_msg_done = true;
516 }
517
518 /*
519  * Prepare headers for the next outgoing message.
520  */
521 static void prepare_write_message(struct ceph_connection *con)
522 {
523         struct ceph_msg *m;
524         u32 crc;
525
526         ceph_con_out_kvec_reset(con);
527         con->out_kvec_is_msg = true;
528         con->out_msg_done = false;
529
530         /* Sneak an ack in there first?  If we can get it into the same
531          * TCP packet that's a good thing. */
532         if (con->in_seq > con->in_seq_acked) {
533                 con->in_seq_acked = con->in_seq;
534                 ceph_con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
535                 con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
536                 ceph_con_out_kvec_add(con, sizeof (con->out_temp_ack),
537                         &con->out_temp_ack);
538         }
539
540         m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
541         con->out_msg = m;
542
543         /* put message on sent list */
544         ceph_msg_get(m);
545         list_move_tail(&m->list_head, &con->out_sent);
546
547         /*
548          * only assign outgoing seq # if we haven't sent this message
549          * yet.  if it is requeued, resend with it's original seq.
550          */
551         if (m->needs_out_seq) {
552                 m->hdr.seq = cpu_to_le64(++con->out_seq);
553                 m->needs_out_seq = false;
554         }
555
556         dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
557              m, con->out_seq, le16_to_cpu(m->hdr.type),
558              le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
559              le32_to_cpu(m->hdr.data_len),
560              m->nr_pages);
561         BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);
562
563         /* tag + hdr + front + middle */
564         ceph_con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
565         ceph_con_out_kvec_add(con, sizeof (m->hdr), &m->hdr);
566         ceph_con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
567
568         if (m->middle)
569                 ceph_con_out_kvec_add(con, m->middle->vec.iov_len,
570                         m->middle->vec.iov_base);
571
572         /* fill in crc (except data pages), footer */
573         crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
574         con->out_msg->hdr.crc = cpu_to_le32(crc);
575         con->out_msg->footer.flags = CEPH_MSG_FOOTER_COMPLETE;
576
577         crc = crc32c(0, m->front.iov_base, m->front.iov_len);
578         con->out_msg->footer.front_crc = cpu_to_le32(crc);
579         if (m->middle) {
580                 crc = crc32c(0, m->middle->vec.iov_base,
581                                 m->middle->vec.iov_len);
582                 con->out_msg->footer.middle_crc = cpu_to_le32(crc);
583         } else
584                 con->out_msg->footer.middle_crc = 0;
585         con->out_msg->footer.data_crc = 0;
586         dout("prepare_write_message front_crc %u data_crc %u\n",
587              le32_to_cpu(con->out_msg->footer.front_crc),
588              le32_to_cpu(con->out_msg->footer.middle_crc));
589
590         /* is there a data payload? */
591         if (le32_to_cpu(m->hdr.data_len) > 0) {
592                 /* initialize page iterator */
593                 con->out_msg_pos.page = 0;
594                 if (m->pages)
595                         con->out_msg_pos.page_pos = m->page_alignment;
596                 else
597                         con->out_msg_pos.page_pos = 0;
598                 con->out_msg_pos.data_pos = 0;
599                 con->out_msg_pos.did_page_crc = false;
600                 con->out_more = 1;  /* data + footer will follow */
601         } else {
602                 /* no, queue up footer too and be done */
603                 prepare_write_message_footer(con);
604         }
605
606         set_bit(WRITE_PENDING, &con->state);
607 }
608
609 /*
610  * Prepare an ack.
611  */
612 static void prepare_write_ack(struct ceph_connection *con)
613 {
614         dout("prepare_write_ack %p %llu -> %llu\n", con,
615              con->in_seq_acked, con->in_seq);
616         con->in_seq_acked = con->in_seq;
617
618         ceph_con_out_kvec_reset(con);
619
620         ceph_con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
621
622         con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
623         ceph_con_out_kvec_add(con, sizeof (con->out_temp_ack),
624                                 &con->out_temp_ack);
625
626         con->out_more = 1;  /* more will follow.. eventually.. */
627         set_bit(WRITE_PENDING, &con->state);
628 }
629
630 /*
631  * Prepare to write keepalive byte.
632  */
633 static void prepare_write_keepalive(struct ceph_connection *con)
634 {
635         dout("prepare_write_keepalive %p\n", con);
636         ceph_con_out_kvec_reset(con);
637         ceph_con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
638         set_bit(WRITE_PENDING, &con->state);
639 }
640
641 /*
642  * Connection negotiation.
643  */
644
645 static int prepare_connect_authorizer(struct ceph_connection *con)
646 {
647         void *auth_buf;
648         int auth_len = 0;
649         int auth_protocol = 0;
650
651         mutex_unlock(&con->mutex);
652         if (con->ops->get_authorizer)
653                 con->ops->get_authorizer(con, &auth_buf, &auth_len,
654                                          &auth_protocol, &con->auth_reply_buf,
655                                          &con->auth_reply_buf_len,
656                                          con->auth_retry);
657         mutex_lock(&con->mutex);
658
659         if (test_bit(CLOSED, &con->state) ||
660             test_bit(OPENING, &con->state))
661                 return -EAGAIN;
662
663         con->out_connect.authorizer_protocol = cpu_to_le32(auth_protocol);
664         con->out_connect.authorizer_len = cpu_to_le32(auth_len);
665
666         if (auth_len)
667                 ceph_con_out_kvec_add(con, auth_len, auth_buf);
668
669         return 0;
670 }
671
672 /*
673  * We connected to a peer and are saying hello.
674  */
675 static void prepare_write_banner(struct ceph_messenger *msgr,
676                                  struct ceph_connection *con)
677 {
678         ceph_con_out_kvec_reset(con);
679         ceph_con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
680         ceph_con_out_kvec_add(con, sizeof (msgr->my_enc_addr),
681                                         &msgr->my_enc_addr);
682
683         con->out_more = 0;
684         set_bit(WRITE_PENDING, &con->state);
685 }
686
687 static int prepare_write_connect(struct ceph_messenger *msgr,
688                                  struct ceph_connection *con,
689                                  int include_banner)
690 {
691         unsigned global_seq = get_global_seq(con->msgr, 0);
692         int proto;
693
694         switch (con->peer_name.type) {
695         case CEPH_ENTITY_TYPE_MON:
696                 proto = CEPH_MONC_PROTOCOL;
697                 break;
698         case CEPH_ENTITY_TYPE_OSD:
699                 proto = CEPH_OSDC_PROTOCOL;
700                 break;
701         case CEPH_ENTITY_TYPE_MDS:
702                 proto = CEPH_MDSC_PROTOCOL;
703                 break;
704         default:
705                 BUG();
706         }
707
708         dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
709              con->connect_seq, global_seq, proto);
710
711         con->out_connect.features = cpu_to_le64(msgr->supported_features);
712         con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
713         con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
714         con->out_connect.global_seq = cpu_to_le32(global_seq);
715         con->out_connect.protocol_version = cpu_to_le32(proto);
716         con->out_connect.flags = 0;
717
718         if (include_banner)
719                 prepare_write_banner(msgr, con);
720         else
721                 ceph_con_out_kvec_reset(con);
722         ceph_con_out_kvec_add(con, sizeof (con->out_connect), &con->out_connect);
723
724         con->out_more = 0;
725         set_bit(WRITE_PENDING, &con->state);
726
727         return prepare_connect_authorizer(con);
728 }
729
730 /*
731  * write as much of pending kvecs to the socket as we can.
732  *  1 -> done
733  *  0 -> socket full, but more to do
734  * <0 -> error
735  */
736 static int write_partial_kvec(struct ceph_connection *con)
737 {
738         int ret;
739
740         dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
741         while (con->out_kvec_bytes > 0) {
742                 ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
743                                        con->out_kvec_left, con->out_kvec_bytes,
744                                        con->out_more);
745                 if (ret <= 0)
746                         goto out;
747                 con->out_kvec_bytes -= ret;
748                 if (con->out_kvec_bytes == 0)
749                         break;            /* done */
750
751                 /* account for full iov entries consumed */
752                 while (ret >= con->out_kvec_cur->iov_len) {
753                         BUG_ON(!con->out_kvec_left);
754                         ret -= con->out_kvec_cur->iov_len;
755                         con->out_kvec_cur++;
756                         con->out_kvec_left--;
757                 }
758                 /* and for a partially-consumed entry */
759                 if (ret) {
760                         con->out_kvec_cur->iov_len -= ret;
761                         con->out_kvec_cur->iov_base += ret;
762                 }
763         }
764         con->out_kvec_left = 0;
765         con->out_kvec_is_msg = false;
766         ret = 1;
767 out:
768         dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
769              con->out_kvec_bytes, con->out_kvec_left, ret);
770         return ret;  /* done! */
771 }
772
773 #ifdef CONFIG_BLOCK
774 static void init_bio_iter(struct bio *bio, struct bio **iter, int *seg)
775 {
776         if (!bio) {
777                 *iter = NULL;
778                 *seg = 0;
779                 return;
780         }
781         *iter = bio;
782         *seg = bio->bi_idx;
783 }
784
785 static void iter_bio_next(struct bio **bio_iter, int *seg)
786 {
787         if (*bio_iter == NULL)
788                 return;
789
790         BUG_ON(*seg >= (*bio_iter)->bi_vcnt);
791
792         (*seg)++;
793         if (*seg == (*bio_iter)->bi_vcnt)
794                 init_bio_iter((*bio_iter)->bi_next, bio_iter, seg);
795 }
796 #endif
797
798 /*
799  * Write as much message data payload as we can.  If we finish, queue
800  * up the footer.
801  *  1 -> done, footer is now queued in out_kvec[].
802  *  0 -> socket full, but more to do
803  * <0 -> error
804  */
805 static int write_partial_msg_pages(struct ceph_connection *con)
806 {
807         struct ceph_msg *msg = con->out_msg;
808         unsigned data_len = le32_to_cpu(msg->hdr.data_len);
809         size_t len;
810         bool do_crc = con->msgr->nocrc;
811         int ret;
812         int total_max_write;
813         int in_trail = 0;
814         size_t trail_len = (msg->trail ? msg->trail->length : 0);
815
816         dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
817              con, con->out_msg, con->out_msg_pos.page, con->out_msg->nr_pages,
818              con->out_msg_pos.page_pos);
819
820 #ifdef CONFIG_BLOCK
821         if (msg->bio && !msg->bio_iter)
822                 init_bio_iter(msg->bio, &msg->bio_iter, &msg->bio_seg);
823 #endif
824
825         while (data_len > con->out_msg_pos.data_pos) {
826                 struct page *page = NULL;
827                 void *kaddr = NULL;
828                 int max_write = PAGE_SIZE;
829                 int page_shift = 0;
830
831                 total_max_write = data_len - trail_len -
832                         con->out_msg_pos.data_pos;
833
834                 /*
835                  * if we are calculating the data crc (the default), we need
836                  * to map the page.  if our pages[] has been revoked, use the
837                  * zero page.
838                  */
839
840                 /* have we reached the trail part of the data? */
841                 if (con->out_msg_pos.data_pos >= data_len - trail_len) {
842                         in_trail = 1;
843
844                         total_max_write = data_len - con->out_msg_pos.data_pos;
845
846                         page = list_first_entry(&msg->trail->head,
847                                                 struct page, lru);
848                         if (do_crc)
849                                 kaddr = kmap(page);
850                         max_write = PAGE_SIZE;
851                 } else if (msg->pages) {
852                         page = msg->pages[con->out_msg_pos.page];
853                         if (do_crc)
854                                 kaddr = kmap(page);
855                 } else if (msg->pagelist) {
856                         page = list_first_entry(&msg->pagelist->head,
857                                                 struct page, lru);
858                         if (do_crc)
859                                 kaddr = kmap(page);
860 #ifdef CONFIG_BLOCK
861                 } else if (msg->bio) {
862                         struct bio_vec *bv;
863
864                         bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
865                         page = bv->bv_page;
866                         page_shift = bv->bv_offset;
867                         if (do_crc)
868                                 kaddr = kmap(page) + page_shift;
869                         max_write = bv->bv_len;
870 #endif
871                 } else {
872                         page = zero_page;
873                         if (do_crc)
874                                 kaddr = zero_page_address;
875                 }
876                 len = min_t(int, max_write - con->out_msg_pos.page_pos,
877                             total_max_write);
878
879                 if (do_crc && !con->out_msg_pos.did_page_crc) {
880                         u32 crc;
881                         void *base = kaddr + con->out_msg_pos.page_pos;
882                         u32 tmpcrc = le32_to_cpu(con->out_msg->footer.data_crc);
883
884                         BUG_ON(kaddr == NULL);
885                         crc = crc32c(tmpcrc, base, len);
886                         con->out_msg->footer.data_crc = cpu_to_le32(crc);
887                         con->out_msg_pos.did_page_crc = true;
888                 }
889                 ret = kernel_sendpage(con->sock, page,
890                                       con->out_msg_pos.page_pos + page_shift,
891                                       len,
892                                       MSG_DONTWAIT | MSG_NOSIGNAL |
893                                       MSG_MORE);
894
895                 if (do_crc &&
896                     (msg->pages || msg->pagelist || msg->bio || in_trail))
897                         kunmap(page);
898
899                 if (ret == -EAGAIN)
900                         ret = 0;
901                 if (ret <= 0)
902                         goto out;
903
904                 con->out_msg_pos.data_pos += ret;
905                 con->out_msg_pos.page_pos += ret;
906                 if (ret == len) {
907                         con->out_msg_pos.page_pos = 0;
908                         con->out_msg_pos.page++;
909                         con->out_msg_pos.did_page_crc = false;
910                         if (in_trail)
911                                 list_move_tail(&page->lru,
912                                                &msg->trail->head);
913                         else if (msg->pagelist)
914                                 list_move_tail(&page->lru,
915                                                &msg->pagelist->head);
916 #ifdef CONFIG_BLOCK
917                         else if (msg->bio)
918                                 iter_bio_next(&msg->bio_iter, &msg->bio_seg);
919 #endif
920                 }
921         }
922
923         dout("write_partial_msg_pages %p msg %p done\n", con, msg);
924
925         /* prepare and queue up footer, too */
926         if (!do_crc)
927                 con->out_msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
928         ceph_con_out_kvec_reset(con);
929         prepare_write_message_footer(con);
930         ret = 1;
931 out:
932         return ret;
933 }
934
935 /*
936  * write some zeros
937  */
938 static int write_partial_skip(struct ceph_connection *con)
939 {
940         int ret;
941
942         while (con->out_skip > 0) {
943                 struct kvec iov = {
944                         .iov_base = zero_page_address,
945                         .iov_len = min(con->out_skip, (int)PAGE_CACHE_SIZE)
946                 };
947
948                 ret = ceph_tcp_sendmsg(con->sock, &iov, 1, iov.iov_len, 1);
949                 if (ret <= 0)
950                         goto out;
951                 con->out_skip -= ret;
952         }
953         ret = 1;
954 out:
955         return ret;
956 }
957
958 /*
959  * Prepare to read connection handshake, or an ack.
960  */
961 static void prepare_read_banner(struct ceph_connection *con)
962 {
963         dout("prepare_read_banner %p\n", con);
964         con->in_base_pos = 0;
965 }
966
967 static void prepare_read_connect(struct ceph_connection *con)
968 {
969         dout("prepare_read_connect %p\n", con);
970         con->in_base_pos = 0;
971 }
972
973 static void prepare_read_ack(struct ceph_connection *con)
974 {
975         dout("prepare_read_ack %p\n", con);
976         con->in_base_pos = 0;
977 }
978
979 static void prepare_read_tag(struct ceph_connection *con)
980 {
981         dout("prepare_read_tag %p\n", con);
982         con->in_base_pos = 0;
983         con->in_tag = CEPH_MSGR_TAG_READY;
984 }
985
986 /*
987  * Prepare to read a message.
988  */
989 static int prepare_read_message(struct ceph_connection *con)
990 {
991         dout("prepare_read_message %p\n", con);
992         BUG_ON(con->in_msg != NULL);
993         con->in_base_pos = 0;
994         con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
995         return 0;
996 }
997
998
999 static int read_partial(struct ceph_connection *con,
1000                         int *to, int size, void *object)
1001 {
1002         *to += size;
1003         while (con->in_base_pos < *to) {
1004                 int left = *to - con->in_base_pos;
1005                 int have = size - left;
1006                 int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
1007                 if (ret <= 0)
1008                         return ret;
1009                 con->in_base_pos += ret;
1010         }
1011         return 1;
1012 }
1013
1014
1015 /*
1016  * Read all or part of the connect-side handshake on a new connection
1017  */
1018 static int read_partial_banner(struct ceph_connection *con)
1019 {
1020         int ret, to = 0;
1021
1022         dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
1023
1024         /* peer's banner */
1025         ret = read_partial(con, &to, strlen(CEPH_BANNER), con->in_banner);
1026         if (ret <= 0)
1027                 goto out;
1028         ret = read_partial(con, &to, sizeof(con->actual_peer_addr),
1029                            &con->actual_peer_addr);
1030         if (ret <= 0)
1031                 goto out;
1032         ret = read_partial(con, &to, sizeof(con->peer_addr_for_me),
1033                            &con->peer_addr_for_me);
1034         if (ret <= 0)
1035                 goto out;
1036 out:
1037         return ret;
1038 }
1039
1040 static int read_partial_connect(struct ceph_connection *con)
1041 {
1042         int ret, to = 0;
1043
1044         dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
1045
1046         ret = read_partial(con, &to, sizeof(con->in_reply), &con->in_reply);
1047         if (ret <= 0)
1048                 goto out;
1049         ret = read_partial(con, &to, le32_to_cpu(con->in_reply.authorizer_len),
1050                            con->auth_reply_buf);
1051         if (ret <= 0)
1052                 goto out;
1053
1054         dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
1055              con, (int)con->in_reply.tag,
1056              le32_to_cpu(con->in_reply.connect_seq),
1057              le32_to_cpu(con->in_reply.global_seq));
1058 out:
1059         return ret;
1060
1061 }
1062
1063 /*
1064  * Verify the hello banner looks okay.
1065  */
1066 static int verify_hello(struct ceph_connection *con)
1067 {
1068         if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1069                 pr_err("connect to %s got bad banner\n",
1070                        ceph_pr_addr(&con->peer_addr.in_addr));
1071                 con->error_msg = "protocol error, bad banner";
1072                 return -1;
1073         }
1074         return 0;
1075 }
1076
1077 static bool addr_is_blank(struct sockaddr_storage *ss)
1078 {
1079         switch (ss->ss_family) {
1080         case AF_INET:
1081                 return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
1082         case AF_INET6:
1083                 return
1084                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
1085                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
1086                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
1087                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
1088         }
1089         return false;
1090 }
1091
1092 static int addr_port(struct sockaddr_storage *ss)
1093 {
1094         switch (ss->ss_family) {
1095         case AF_INET:
1096                 return ntohs(((struct sockaddr_in *)ss)->sin_port);
1097         case AF_INET6:
1098                 return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
1099         }
1100         return 0;
1101 }
1102
1103 static void addr_set_port(struct sockaddr_storage *ss, int p)
1104 {
1105         switch (ss->ss_family) {
1106         case AF_INET:
1107                 ((struct sockaddr_in *)ss)->sin_port = htons(p);
1108                 break;
1109         case AF_INET6:
1110                 ((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1111                 break;
1112         }
1113 }
1114
1115 /*
1116  * Unlike other *_pton function semantics, zero indicates success.
1117  */
1118 static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
1119                 char delim, const char **ipend)
1120 {
1121         struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
1122         struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
1123
1124         memset(ss, 0, sizeof(*ss));
1125
1126         if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
1127                 ss->ss_family = AF_INET;
1128                 return 0;
1129         }
1130
1131         if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
1132                 ss->ss_family = AF_INET6;
1133                 return 0;
1134         }
1135
1136         return -EINVAL;
1137 }
1138
1139 /*
1140  * Extract hostname string and resolve using kernel DNS facility.
1141  */
1142 #ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
1143 static int ceph_dns_resolve_name(const char *name, size_t namelen,
1144                 struct sockaddr_storage *ss, char delim, const char **ipend)
1145 {
1146         const char *end, *delim_p;
1147         char *colon_p, *ip_addr = NULL;
1148         int ip_len, ret;
1149
1150         /*
1151          * The end of the hostname occurs immediately preceding the delimiter or
1152          * the port marker (':') where the delimiter takes precedence.
1153          */
1154         delim_p = memchr(name, delim, namelen);
1155         colon_p = memchr(name, ':', namelen);
1156
1157         if (delim_p && colon_p)
1158                 end = delim_p < colon_p ? delim_p : colon_p;
1159         else if (!delim_p && colon_p)
1160                 end = colon_p;
1161         else {
1162                 end = delim_p;
1163                 if (!end) /* case: hostname:/ */
1164                         end = name + namelen;
1165         }
1166
1167         if (end <= name)
1168                 return -EINVAL;
1169
1170         /* do dns_resolve upcall */
1171         ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
1172         if (ip_len > 0)
1173                 ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
1174         else
1175                 ret = -ESRCH;
1176
1177         kfree(ip_addr);
1178
1179         *ipend = end;
1180
1181         pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
1182                         ret, ret ? "failed" : ceph_pr_addr(ss));
1183
1184         return ret;
1185 }
1186 #else
1187 static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
1188                 struct sockaddr_storage *ss, char delim, const char **ipend)
1189 {
1190         return -EINVAL;
1191 }
1192 #endif
1193
1194 /*
1195  * Parse a server name (IP or hostname). If a valid IP address is not found
1196  * then try to extract a hostname to resolve using userspace DNS upcall.
1197  */
1198 static int ceph_parse_server_name(const char *name, size_t namelen,
1199                         struct sockaddr_storage *ss, char delim, const char **ipend)
1200 {
1201         int ret;
1202
1203         ret = ceph_pton(name, namelen, ss, delim, ipend);
1204         if (ret)
1205                 ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);
1206
1207         return ret;
1208 }
1209
1210 /*
1211  * Parse an ip[:port] list into an addr array.  Use the default
1212  * monitor port if a port isn't specified.
1213  */
1214 int ceph_parse_ips(const char *c, const char *end,
1215                    struct ceph_entity_addr *addr,
1216                    int max_count, int *count)
1217 {
1218         int i, ret = -EINVAL;
1219         const char *p = c;
1220
1221         dout("parse_ips on '%.*s'\n", (int)(end-c), c);
1222         for (i = 0; i < max_count; i++) {
1223                 const char *ipend;
1224                 struct sockaddr_storage *ss = &addr[i].in_addr;
1225                 int port;
1226                 char delim = ',';
1227
1228                 if (*p == '[') {
1229                         delim = ']';
1230                         p++;
1231                 }
1232
1233                 ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
1234                 if (ret)
1235                         goto bad;
1236                 ret = -EINVAL;
1237
1238                 p = ipend;
1239
1240                 if (delim == ']') {
1241                         if (*p != ']') {
1242                                 dout("missing matching ']'\n");
1243                                 goto bad;
1244                         }
1245                         p++;
1246                 }
1247
1248                 /* port? */
1249                 if (p < end && *p == ':') {
1250                         port = 0;
1251                         p++;
1252                         while (p < end && *p >= '0' && *p <= '9') {
1253                                 port = (port * 10) + (*p - '0');
1254                                 p++;
1255                         }
1256                         if (port > 65535 || port == 0)
1257                                 goto bad;
1258                 } else {
1259                         port = CEPH_MON_PORT;
1260                 }
1261
1262                 addr_set_port(ss, port);
1263
1264                 dout("parse_ips got %s\n", ceph_pr_addr(ss));
1265
1266                 if (p == end)
1267                         break;
1268                 if (*p != ',')
1269                         goto bad;
1270                 p++;
1271         }
1272
1273         if (p != end)
1274                 goto bad;
1275
1276         if (count)
1277                 *count = i + 1;
1278         return 0;
1279
1280 bad:
1281         pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
1282         return ret;
1283 }
1284 EXPORT_SYMBOL(ceph_parse_ips);
1285
1286 static int process_banner(struct ceph_connection *con)
1287 {
1288         dout("process_banner on %p\n", con);
1289
1290         if (verify_hello(con) < 0)
1291                 return -1;
1292
1293         ceph_decode_addr(&con->actual_peer_addr);
1294         ceph_decode_addr(&con->peer_addr_for_me);
1295
1296         /*
1297          * Make sure the other end is who we wanted.  note that the other
1298          * end may not yet know their ip address, so if it's 0.0.0.0, give
1299          * them the benefit of the doubt.
1300          */
1301         if (memcmp(&con->peer_addr, &con->actual_peer_addr,
1302                    sizeof(con->peer_addr)) != 0 &&
1303             !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
1304               con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
1305                 pr_warning("wrong peer, want %s/%d, got %s/%d\n",
1306                            ceph_pr_addr(&con->peer_addr.in_addr),
1307                            (int)le32_to_cpu(con->peer_addr.nonce),
1308                            ceph_pr_addr(&con->actual_peer_addr.in_addr),
1309                            (int)le32_to_cpu(con->actual_peer_addr.nonce));
1310                 con->error_msg = "wrong peer at address";
1311                 return -1;
1312         }
1313
1314         /*
1315          * did we learn our address?
1316          */
1317         if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
1318                 int port = addr_port(&con->msgr->inst.addr.in_addr);
1319
1320                 memcpy(&con->msgr->inst.addr.in_addr,
1321                        &con->peer_addr_for_me.in_addr,
1322                        sizeof(con->peer_addr_for_me.in_addr));
1323                 addr_set_port(&con->msgr->inst.addr.in_addr, port);
1324                 encode_my_addr(con->msgr);
1325                 dout("process_banner learned my addr is %s\n",
1326                      ceph_pr_addr(&con->msgr->inst.addr.in_addr));
1327         }
1328
1329         set_bit(NEGOTIATING, &con->state);
1330         prepare_read_connect(con);
1331         return 0;
1332 }
1333
1334 static void fail_protocol(struct ceph_connection *con)
1335 {
1336         reset_connection(con);
1337         set_bit(CLOSED, &con->state);  /* in case there's queued work */
1338
1339         mutex_unlock(&con->mutex);
1340         if (con->ops->bad_proto)
1341                 con->ops->bad_proto(con);
1342         mutex_lock(&con->mutex);
1343 }
1344
1345 static int process_connect(struct ceph_connection *con)
1346 {
1347         u64 sup_feat = con->msgr->supported_features;
1348         u64 req_feat = con->msgr->required_features;
1349         u64 server_feat = le64_to_cpu(con->in_reply.features);
1350         int ret;
1351
1352         dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
1353
1354         switch (con->in_reply.tag) {
1355         case CEPH_MSGR_TAG_FEATURES:
1356                 pr_err("%s%lld %s feature set mismatch,"
1357                        " my %llx < server's %llx, missing %llx\n",
1358                        ENTITY_NAME(con->peer_name),
1359                        ceph_pr_addr(&con->peer_addr.in_addr),
1360                        sup_feat, server_feat, server_feat & ~sup_feat);
1361                 con->error_msg = "missing required protocol features";
1362                 fail_protocol(con);
1363                 return -1;
1364
1365         case CEPH_MSGR_TAG_BADPROTOVER:
1366                 pr_err("%s%lld %s protocol version mismatch,"
1367                        " my %d != server's %d\n",
1368                        ENTITY_NAME(con->peer_name),
1369                        ceph_pr_addr(&con->peer_addr.in_addr),
1370                        le32_to_cpu(con->out_connect.protocol_version),
1371                        le32_to_cpu(con->in_reply.protocol_version));
1372                 con->error_msg = "protocol version mismatch";
1373                 fail_protocol(con);
1374                 return -1;
1375
1376         case CEPH_MSGR_TAG_BADAUTHORIZER:
1377                 con->auth_retry++;
1378                 dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
1379                      con->auth_retry);
1380                 if (con->auth_retry == 2) {
1381                         con->error_msg = "connect authorization failure";
1382                         return -1;
1383                 }
1384                 con->auth_retry = 1;
1385                 ret = prepare_write_connect(con->msgr, con, 0);
1386                 if (ret < 0)
1387                         return ret;
1388                 prepare_read_connect(con);
1389                 break;
1390
1391         case CEPH_MSGR_TAG_RESETSESSION:
1392                 /*
1393                  * If we connected with a large connect_seq but the peer
1394                  * has no record of a session with us (no connection, or
1395                  * connect_seq == 0), they will send RESETSESION to indicate
1396                  * that they must have reset their session, and may have
1397                  * dropped messages.
1398                  */
1399                 dout("process_connect got RESET peer seq %u\n",
1400                      le32_to_cpu(con->in_connect.connect_seq));
1401                 pr_err("%s%lld %s connection reset\n",
1402                        ENTITY_NAME(con->peer_name),
1403                        ceph_pr_addr(&con->peer_addr.in_addr));
1404                 reset_connection(con);
1405                 prepare_write_connect(con->msgr, con, 0);
1406                 prepare_read_connect(con);
1407
1408                 /* Tell ceph about it. */
1409                 mutex_unlock(&con->mutex);
1410                 pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
1411                 if (con->ops->peer_reset)
1412                         con->ops->peer_reset(con);
1413                 mutex_lock(&con->mutex);
1414                 if (test_bit(CLOSED, &con->state) ||
1415                     test_bit(OPENING, &con->state))
1416                         return -EAGAIN;
1417                 break;
1418
1419         case CEPH_MSGR_TAG_RETRY_SESSION:
1420                 /*
1421                  * If we sent a smaller connect_seq than the peer has, try
1422                  * again with a larger value.
1423                  */
1424                 dout("process_connect got RETRY my seq = %u, peer_seq = %u\n",
1425                      le32_to_cpu(con->out_connect.connect_seq),
1426                      le32_to_cpu(con->in_connect.connect_seq));
1427                 con->connect_seq = le32_to_cpu(con->in_connect.connect_seq);
1428                 prepare_write_connect(con->msgr, con, 0);
1429                 prepare_read_connect(con);
1430                 break;
1431
1432         case CEPH_MSGR_TAG_RETRY_GLOBAL:
1433                 /*
1434                  * If we sent a smaller global_seq than the peer has, try
1435                  * again with a larger value.
1436                  */
1437                 dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
1438                      con->peer_global_seq,
1439                      le32_to_cpu(con->in_connect.global_seq));
1440                 get_global_seq(con->msgr,
1441                                le32_to_cpu(con->in_connect.global_seq));
1442                 prepare_write_connect(con->msgr, con, 0);
1443                 prepare_read_connect(con);
1444                 break;
1445
1446         case CEPH_MSGR_TAG_READY:
1447                 if (req_feat & ~server_feat) {
1448                         pr_err("%s%lld %s protocol feature mismatch,"
1449                                " my required %llx > server's %llx, need %llx\n",
1450                                ENTITY_NAME(con->peer_name),
1451                                ceph_pr_addr(&con->peer_addr.in_addr),
1452                                req_feat, server_feat, req_feat & ~server_feat);
1453                         con->error_msg = "missing required protocol features";
1454                         fail_protocol(con);
1455                         return -1;
1456                 }
1457                 clear_bit(CONNECTING, &con->state);
1458                 con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
1459                 con->connect_seq++;
1460                 con->peer_features = server_feat;
1461                 dout("process_connect got READY gseq %d cseq %d (%d)\n",
1462                      con->peer_global_seq,
1463                      le32_to_cpu(con->in_reply.connect_seq),
1464                      con->connect_seq);
1465                 WARN_ON(con->connect_seq !=
1466                         le32_to_cpu(con->in_reply.connect_seq));
1467
1468                 if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
1469                         set_bit(LOSSYTX, &con->state);
1470
1471                 prepare_read_tag(con);
1472                 break;
1473
1474         case CEPH_MSGR_TAG_WAIT:
1475                 /*
1476                  * If there is a connection race (we are opening
1477                  * connections to each other), one of us may just have
1478                  * to WAIT.  This shouldn't happen if we are the
1479                  * client.
1480                  */
1481                 pr_err("process_connect got WAIT as client\n");
1482                 con->error_msg = "protocol error, got WAIT as client";
1483                 return -1;
1484
1485         default:
1486                 pr_err("connect protocol error, will retry\n");
1487                 con->error_msg = "protocol error, garbage tag during connect";
1488                 return -1;
1489         }
1490         return 0;
1491 }
1492
1493
1494 /*
1495  * read (part of) an ack
1496  */
1497 static int read_partial_ack(struct ceph_connection *con)
1498 {
1499         int to = 0;
1500
1501         return read_partial(con, &to, sizeof(con->in_temp_ack),
1502                             &con->in_temp_ack);
1503 }
1504
1505
1506 /*
1507  * We can finally discard anything that's been acked.
1508  */
1509 static void process_ack(struct ceph_connection *con)
1510 {
1511         struct ceph_msg *m;
1512         u64 ack = le64_to_cpu(con->in_temp_ack);
1513         u64 seq;
1514
1515         while (!list_empty(&con->out_sent)) {
1516                 m = list_first_entry(&con->out_sent, struct ceph_msg,
1517                                      list_head);
1518                 seq = le64_to_cpu(m->hdr.seq);
1519                 if (seq > ack)
1520                         break;
1521                 dout("got ack for seq %llu type %d at %p\n", seq,
1522                      le16_to_cpu(m->hdr.type), m);
1523                 m->ack_stamp = jiffies;
1524                 ceph_msg_remove(m);
1525         }
1526         prepare_read_tag(con);
1527 }
1528
1529
1530
1531
1532 static int read_partial_message_section(struct ceph_connection *con,
1533                                         struct kvec *section,
1534                                         unsigned int sec_len, u32 *crc)
1535 {
1536         int ret, left;
1537
1538         BUG_ON(!section);
1539
1540         while (section->iov_len < sec_len) {
1541                 BUG_ON(section->iov_base == NULL);
1542                 left = sec_len - section->iov_len;
1543                 ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
1544                                        section->iov_len, left);
1545                 if (ret <= 0)
1546                         return ret;
1547                 section->iov_len += ret;
1548         }
1549         if (section->iov_len == sec_len)
1550                 *crc = crc32c(0, section->iov_base, section->iov_len);
1551
1552         return 1;
1553 }
1554
1555 static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
1556                                 struct ceph_msg_header *hdr,
1557                                 int *skip);
1558
1559
1560 static int read_partial_message_pages(struct ceph_connection *con,
1561                                       struct page **pages,
1562                                       unsigned data_len, bool do_datacrc)
1563 {
1564         void *p;
1565         int ret;
1566         int left;
1567
1568         left = min((int)(data_len - con->in_msg_pos.data_pos),
1569                    (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
1570         /* (page) data */
1571         BUG_ON(pages == NULL);
1572         p = kmap(pages[con->in_msg_pos.page]);
1573         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1574                                left);
1575         if (ret > 0 && do_datacrc)
1576                 con->in_data_crc =
1577                         crc32c(con->in_data_crc,
1578                                   p + con->in_msg_pos.page_pos, ret);
1579         kunmap(pages[con->in_msg_pos.page]);
1580         if (ret <= 0)
1581                 return ret;
1582         con->in_msg_pos.data_pos += ret;
1583         con->in_msg_pos.page_pos += ret;
1584         if (con->in_msg_pos.page_pos == PAGE_SIZE) {
1585                 con->in_msg_pos.page_pos = 0;
1586                 con->in_msg_pos.page++;
1587         }
1588
1589         return ret;
1590 }
1591
1592 #ifdef CONFIG_BLOCK
1593 static int read_partial_message_bio(struct ceph_connection *con,
1594                                     struct bio **bio_iter, int *bio_seg,
1595                                     unsigned data_len, bool do_datacrc)
1596 {
1597         struct bio_vec *bv = bio_iovec_idx(*bio_iter, *bio_seg);
1598         void *p;
1599         int ret, left;
1600
1601         if (IS_ERR(bv))
1602                 return PTR_ERR(bv);
1603
1604         left = min((int)(data_len - con->in_msg_pos.data_pos),
1605                    (int)(bv->bv_len - con->in_msg_pos.page_pos));
1606
1607         p = kmap(bv->bv_page) + bv->bv_offset;
1608
1609         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1610                                left);
1611         if (ret > 0 && do_datacrc)
1612                 con->in_data_crc =
1613                         crc32c(con->in_data_crc,
1614                                   p + con->in_msg_pos.page_pos, ret);
1615         kunmap(bv->bv_page);
1616         if (ret <= 0)
1617                 return ret;
1618         con->in_msg_pos.data_pos += ret;
1619         con->in_msg_pos.page_pos += ret;
1620         if (con->in_msg_pos.page_pos == bv->bv_len) {
1621                 con->in_msg_pos.page_pos = 0;
1622                 iter_bio_next(bio_iter, bio_seg);
1623         }
1624
1625         return ret;
1626 }
1627 #endif
1628
1629 /*
1630  * read (part of) a message.
1631  */
1632 static int read_partial_message(struct ceph_connection *con)
1633 {
1634         struct ceph_msg *m = con->in_msg;
1635         int ret;
1636         int to, left;
1637         unsigned front_len, middle_len, data_len;
1638         bool do_datacrc = con->msgr->nocrc;
1639         int skip;
1640         u64 seq;
1641         u32 crc;
1642
1643         dout("read_partial_message con %p msg %p\n", con, m);
1644
1645         /* header */
1646         while (con->in_base_pos < sizeof(con->in_hdr)) {
1647                 left = sizeof(con->in_hdr) - con->in_base_pos;
1648                 ret = ceph_tcp_recvmsg(con->sock,
1649                                        (char *)&con->in_hdr + con->in_base_pos,
1650                                        left);
1651                 if (ret <= 0)
1652                         return ret;
1653                 con->in_base_pos += ret;
1654         }
1655
1656         crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
1657         if (cpu_to_le32(crc) != con->in_hdr.crc) {
1658                 pr_err("read_partial_message bad hdr "
1659                        " crc %u != expected %u\n",
1660                        crc, con->in_hdr.crc);
1661                 return -EBADMSG;
1662         }
1663
1664         front_len = le32_to_cpu(con->in_hdr.front_len);
1665         if (front_len > CEPH_MSG_MAX_FRONT_LEN)
1666                 return -EIO;
1667         middle_len = le32_to_cpu(con->in_hdr.middle_len);
1668         if (middle_len > CEPH_MSG_MAX_DATA_LEN)
1669                 return -EIO;
1670         data_len = le32_to_cpu(con->in_hdr.data_len);
1671         if (data_len > CEPH_MSG_MAX_DATA_LEN)
1672                 return -EIO;
1673
1674         /* verify seq# */
1675         seq = le64_to_cpu(con->in_hdr.seq);
1676         if ((s64)seq - (s64)con->in_seq < 1) {
1677                 pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1678                         ENTITY_NAME(con->peer_name),
1679                         ceph_pr_addr(&con->peer_addr.in_addr),
1680                         seq, con->in_seq + 1);
1681                 con->in_base_pos = -front_len - middle_len - data_len -
1682                         sizeof(m->footer);
1683                 con->in_tag = CEPH_MSGR_TAG_READY;
1684                 return 0;
1685         } else if ((s64)seq - (s64)con->in_seq > 1) {
1686                 pr_err("read_partial_message bad seq %lld expected %lld\n",
1687                        seq, con->in_seq + 1);
1688                 con->error_msg = "bad message sequence # for incoming message";
1689                 return -EBADMSG;
1690         }
1691
1692         /* allocate message? */
1693         if (!con->in_msg) {
1694                 dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
1695                      con->in_hdr.front_len, con->in_hdr.data_len);
1696                 skip = 0;
1697                 con->in_msg = ceph_alloc_msg(con, &con->in_hdr, &skip);
1698                 if (skip) {
1699                         /* skip this message */
1700                         dout("alloc_msg said skip message\n");
1701                         BUG_ON(con->in_msg);
1702                         con->in_base_pos = -front_len - middle_len - data_len -
1703                                 sizeof(m->footer);
1704                         con->in_tag = CEPH_MSGR_TAG_READY;
1705                         con->in_seq++;
1706                         return 0;
1707                 }
1708                 if (!con->in_msg) {
1709                         con->error_msg =
1710                                 "error allocating memory for incoming message";
1711                         return -ENOMEM;
1712                 }
1713                 m = con->in_msg;
1714                 m->front.iov_len = 0;    /* haven't read it yet */
1715                 if (m->middle)
1716                         m->middle->vec.iov_len = 0;
1717
1718                 con->in_msg_pos.page = 0;
1719                 if (m->pages)
1720                         con->in_msg_pos.page_pos = m->page_alignment;
1721                 else
1722                         con->in_msg_pos.page_pos = 0;
1723                 con->in_msg_pos.data_pos = 0;
1724         }
1725
1726         /* front */
1727         ret = read_partial_message_section(con, &m->front, front_len,
1728                                            &con->in_front_crc);
1729         if (ret <= 0)
1730                 return ret;
1731
1732         /* middle */
1733         if (m->middle) {
1734                 ret = read_partial_message_section(con, &m->middle->vec,
1735                                                    middle_len,
1736                                                    &con->in_middle_crc);
1737                 if (ret <= 0)
1738                         return ret;
1739         }
1740 #ifdef CONFIG_BLOCK
1741         if (m->bio && !m->bio_iter)
1742                 init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
1743 #endif
1744
1745         /* (page) data */
1746         while (con->in_msg_pos.data_pos < data_len) {
1747                 if (m->pages) {
1748                         ret = read_partial_message_pages(con, m->pages,
1749                                                  data_len, do_datacrc);
1750                         if (ret <= 0)
1751                                 return ret;
1752 #ifdef CONFIG_BLOCK
1753                 } else if (m->bio) {
1754
1755                         ret = read_partial_message_bio(con,
1756                                                  &m->bio_iter, &m->bio_seg,
1757                                                  data_len, do_datacrc);
1758                         if (ret <= 0)
1759                                 return ret;
1760 #endif
1761                 } else {
1762                         BUG_ON(1);
1763                 }
1764         }
1765
1766         /* footer */
1767         to = sizeof(m->hdr) + sizeof(m->footer);
1768         while (con->in_base_pos < to) {
1769                 left = to - con->in_base_pos;
1770                 ret = ceph_tcp_recvmsg(con->sock, (char *)&m->footer +
1771                                        (con->in_base_pos - sizeof(m->hdr)),
1772                                        left);
1773                 if (ret <= 0)
1774                         return ret;
1775                 con->in_base_pos += ret;
1776         }
1777         dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
1778              m, front_len, m->footer.front_crc, middle_len,
1779              m->footer.middle_crc, data_len, m->footer.data_crc);
1780
1781         /* crc ok? */
1782         if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
1783                 pr_err("read_partial_message %p front crc %u != exp. %u\n",
1784                        m, con->in_front_crc, m->footer.front_crc);
1785                 return -EBADMSG;
1786         }
1787         if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
1788                 pr_err("read_partial_message %p middle crc %u != exp %u\n",
1789                        m, con->in_middle_crc, m->footer.middle_crc);
1790                 return -EBADMSG;
1791         }
1792         if (do_datacrc &&
1793             (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
1794             con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
1795                 pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
1796                        con->in_data_crc, le32_to_cpu(m->footer.data_crc));
1797                 return -EBADMSG;
1798         }
1799
1800         return 1; /* done! */
1801 }
1802
1803 /*
1804  * Process message.  This happens in the worker thread.  The callback should
1805  * be careful not to do anything that waits on other incoming messages or it
1806  * may deadlock.
1807  */
1808 static void process_message(struct ceph_connection *con)
1809 {
1810         struct ceph_msg *msg;
1811
1812         msg = con->in_msg;
1813         con->in_msg = NULL;
1814
1815         /* if first message, set peer_name */
1816         if (con->peer_name.type == 0)
1817                 con->peer_name = msg->hdr.src;
1818
1819         con->in_seq++;
1820         mutex_unlock(&con->mutex);
1821
1822         dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
1823              msg, le64_to_cpu(msg->hdr.seq),
1824              ENTITY_NAME(msg->hdr.src),
1825              le16_to_cpu(msg->hdr.type),
1826              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
1827              le32_to_cpu(msg->hdr.front_len),
1828              le32_to_cpu(msg->hdr.data_len),
1829              con->in_front_crc, con->in_middle_crc, con->in_data_crc);
1830         con->ops->dispatch(con, msg);
1831
1832         mutex_lock(&con->mutex);
1833         prepare_read_tag(con);
1834 }
1835
1836
1837 /*
1838  * Write something to the socket.  Called in a worker thread when the
1839  * socket appears to be writeable and we have something ready to send.
1840  */
1841 static int try_write(struct ceph_connection *con)
1842 {
1843         struct ceph_messenger *msgr = con->msgr;
1844         int ret = 1;
1845
1846         dout("try_write start %p state %lu nref %d\n", con, con->state,
1847              atomic_read(&con->nref));
1848
1849 more:
1850         dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
1851
1852         /* open the socket first? */
1853         if (con->sock == NULL) {
1854                 prepare_write_connect(msgr, con, 1);
1855                 prepare_read_banner(con);
1856                 set_bit(CONNECTING, &con->state);
1857                 clear_bit(NEGOTIATING, &con->state);
1858
1859                 BUG_ON(con->in_msg);
1860                 con->in_tag = CEPH_MSGR_TAG_READY;
1861                 dout("try_write initiating connect on %p new state %lu\n",
1862                      con, con->state);
1863                 ret = ceph_tcp_connect(con);
1864                 if (ret < 0) {
1865                         con->error_msg = "connect error";
1866                         goto out;
1867                 }
1868         }
1869
1870 more_kvec:
1871         /* kvec data queued? */
1872         if (con->out_skip) {
1873                 ret = write_partial_skip(con);
1874                 if (ret <= 0)
1875                         goto out;
1876         }
1877         if (con->out_kvec_left) {
1878                 ret = write_partial_kvec(con);
1879                 if (ret <= 0)
1880                         goto out;
1881         }
1882
1883         /* msg pages? */
1884         if (con->out_msg) {
1885                 if (con->out_msg_done) {
1886                         ceph_msg_put(con->out_msg);
1887                         con->out_msg = NULL;   /* we're done with this one */
1888                         goto do_next;
1889                 }
1890
1891                 ret = write_partial_msg_pages(con);
1892                 if (ret == 1)
1893                         goto more_kvec;  /* we need to send the footer, too! */
1894                 if (ret == 0)
1895                         goto out;
1896                 if (ret < 0) {
1897                         dout("try_write write_partial_msg_pages err %d\n",
1898                              ret);
1899                         goto out;
1900                 }
1901         }
1902
1903 do_next:
1904         if (!test_bit(CONNECTING, &con->state)) {
1905                 /* is anything else pending? */
1906                 if (!list_empty(&con->out_queue)) {
1907                         prepare_write_message(con);
1908                         goto more;
1909                 }
1910                 if (con->in_seq > con->in_seq_acked) {
1911                         prepare_write_ack(con);
1912                         goto more;
1913                 }
1914                 if (test_and_clear_bit(KEEPALIVE_PENDING, &con->state)) {
1915                         prepare_write_keepalive(con);
1916                         goto more;
1917                 }
1918         }
1919
1920         /* Nothing to do! */
1921         clear_bit(WRITE_PENDING, &con->state);
1922         dout("try_write nothing else to write.\n");
1923         ret = 0;
1924 out:
1925         dout("try_write done on %p ret %d\n", con, ret);
1926         return ret;
1927 }
1928
1929
1930
1931 /*
1932  * Read what we can from the socket.
1933  */
1934 static int try_read(struct ceph_connection *con)
1935 {
1936         int ret = -1;
1937
1938         if (!con->sock)
1939                 return 0;
1940
1941         if (test_bit(STANDBY, &con->state))
1942                 return 0;
1943
1944         dout("try_read start on %p\n", con);
1945
1946 more:
1947         dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
1948              con->in_base_pos);
1949
1950         /*
1951          * process_connect and process_message drop and re-take
1952          * con->mutex.  make sure we handle a racing close or reopen.
1953          */
1954         if (test_bit(CLOSED, &con->state) ||
1955             test_bit(OPENING, &con->state)) {
1956                 ret = -EAGAIN;
1957                 goto out;
1958         }
1959
1960         if (test_bit(CONNECTING, &con->state)) {
1961                 if (!test_bit(NEGOTIATING, &con->state)) {
1962                         dout("try_read connecting\n");
1963                         ret = read_partial_banner(con);
1964                         if (ret <= 0)
1965                                 goto out;
1966                         ret = process_banner(con);
1967                         if (ret < 0)
1968                                 goto out;
1969                 }
1970                 ret = read_partial_connect(con);
1971                 if (ret <= 0)
1972                         goto out;
1973                 ret = process_connect(con);
1974                 if (ret < 0)
1975                         goto out;
1976                 goto more;
1977         }
1978
1979         if (con->in_base_pos < 0) {
1980                 /*
1981                  * skipping + discarding content.
1982                  *
1983                  * FIXME: there must be a better way to do this!
1984                  */
1985                 static char buf[1024];
1986                 int skip = min(1024, -con->in_base_pos);
1987                 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
1988                 ret = ceph_tcp_recvmsg(con->sock, buf, skip);
1989                 if (ret <= 0)
1990                         goto out;
1991                 con->in_base_pos += ret;
1992                 if (con->in_base_pos)
1993                         goto more;
1994         }
1995         if (con->in_tag == CEPH_MSGR_TAG_READY) {
1996                 /*
1997                  * what's next?
1998                  */
1999                 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
2000                 if (ret <= 0)
2001                         goto out;
2002                 dout("try_read got tag %d\n", (int)con->in_tag);
2003                 switch (con->in_tag) {
2004                 case CEPH_MSGR_TAG_MSG:
2005                         prepare_read_message(con);
2006                         break;
2007                 case CEPH_MSGR_TAG_ACK:
2008                         prepare_read_ack(con);
2009                         break;
2010                 case CEPH_MSGR_TAG_CLOSE:
2011                         set_bit(CLOSED, &con->state);   /* fixme */
2012                         goto out;
2013                 default:
2014                         goto bad_tag;
2015                 }
2016         }
2017         if (con->in_tag == CEPH_MSGR_TAG_MSG) {
2018                 ret = read_partial_message(con);
2019                 if (ret <= 0) {
2020                         switch (ret) {
2021                         case -EBADMSG:
2022                                 con->error_msg = "bad crc";
2023                                 ret = -EIO;
2024                                 break;
2025                         case -EIO:
2026                                 con->error_msg = "io error";
2027                                 break;
2028                         }
2029                         goto out;
2030                 }
2031                 if (con->in_tag == CEPH_MSGR_TAG_READY)
2032                         goto more;
2033                 process_message(con);
2034                 goto more;
2035         }
2036         if (con->in_tag == CEPH_MSGR_TAG_ACK) {
2037                 ret = read_partial_ack(con);
2038                 if (ret <= 0)
2039                         goto out;
2040                 process_ack(con);
2041                 goto more;
2042         }
2043
2044 out:
2045         dout("try_read done on %p ret %d\n", con, ret);
2046         return ret;
2047
2048 bad_tag:
2049         pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
2050         con->error_msg = "protocol error, garbage tag";
2051         ret = -1;
2052         goto out;
2053 }
2054
2055
2056 /*
2057  * Atomically queue work on a connection.  Bump @con reference to
2058  * avoid races with connection teardown.
2059  */
2060 static void queue_con(struct ceph_connection *con)
2061 {
2062         if (test_bit(DEAD, &con->state)) {
2063                 dout("queue_con %p ignoring: DEAD\n",
2064                      con);
2065                 return;
2066         }
2067
2068         if (!con->ops->get(con)) {
2069                 dout("queue_con %p ref count 0\n", con);
2070                 return;
2071         }
2072
2073         if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
2074                 dout("queue_con %p - already queued\n", con);
2075                 con->ops->put(con);
2076         } else {
2077                 dout("queue_con %p\n", con);
2078         }
2079 }
2080
2081 /*
2082  * Do some work on a connection.  Drop a connection ref when we're done.
2083  */
2084 static void con_work(struct work_struct *work)
2085 {
2086         struct ceph_connection *con = container_of(work, struct ceph_connection,
2087                                                    work.work);
2088         int ret;
2089
2090         mutex_lock(&con->mutex);
2091 restart:
2092         if (test_and_clear_bit(BACKOFF, &con->state)) {
2093                 dout("con_work %p backing off\n", con);
2094                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2095                                        round_jiffies_relative(con->delay))) {
2096                         dout("con_work %p backoff %lu\n", con, con->delay);
2097                         mutex_unlock(&con->mutex);
2098                         return;
2099                 } else {
2100                         con->ops->put(con);
2101                         dout("con_work %p FAILED to back off %lu\n", con,
2102                              con->delay);
2103                 }
2104         }
2105
2106         if (test_bit(STANDBY, &con->state)) {
2107                 dout("con_work %p STANDBY\n", con);
2108                 goto done;
2109         }
2110         if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
2111                 dout("con_work CLOSED\n");
2112                 con_close_socket(con);
2113                 goto done;
2114         }
2115         if (test_and_clear_bit(OPENING, &con->state)) {
2116                 /* reopen w/ new peer */
2117                 dout("con_work OPENING\n");
2118                 con_close_socket(con);
2119         }
2120
2121         if (test_and_clear_bit(SOCK_CLOSED, &con->state))
2122                 goto fault;
2123
2124         ret = try_read(con);
2125         if (ret == -EAGAIN)
2126                 goto restart;
2127         if (ret < 0)
2128                 goto fault;
2129
2130         ret = try_write(con);
2131         if (ret == -EAGAIN)
2132                 goto restart;
2133         if (ret < 0)
2134                 goto fault;
2135
2136 done:
2137         mutex_unlock(&con->mutex);
2138 done_unlocked:
2139         con->ops->put(con);
2140         return;
2141
2142 fault:
2143         mutex_unlock(&con->mutex);
2144         ceph_fault(con);     /* error/fault path */
2145         goto done_unlocked;
2146 }
2147
2148
2149 /*
2150  * Generic error/fault handler.  A retry mechanism is used with
2151  * exponential backoff
2152  */
2153 static void ceph_fault(struct ceph_connection *con)
2154 {
2155         pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2156                ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
2157         dout("fault %p state %lu to peer %s\n",
2158              con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
2159
2160         if (test_bit(LOSSYTX, &con->state)) {
2161                 dout("fault on LOSSYTX channel\n");
2162                 goto out;
2163         }
2164
2165         mutex_lock(&con->mutex);
2166         if (test_bit(CLOSED, &con->state))
2167                 goto out_unlock;
2168
2169         con_close_socket(con);
2170
2171         if (con->in_msg) {
2172                 ceph_msg_put(con->in_msg);
2173                 con->in_msg = NULL;
2174         }
2175
2176         /* Requeue anything that hasn't been acked */
2177         list_splice_init(&con->out_sent, &con->out_queue);
2178
2179         /* If there are no messages queued or keepalive pending, place
2180          * the connection in a STANDBY state */
2181         if (list_empty(&con->out_queue) &&
2182             !test_bit(KEEPALIVE_PENDING, &con->state)) {
2183                 dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2184                 clear_bit(WRITE_PENDING, &con->state);
2185                 set_bit(STANDBY, &con->state);
2186         } else {
2187                 /* retry after a delay. */
2188                 if (con->delay == 0)
2189                         con->delay = BASE_DELAY_INTERVAL;
2190                 else if (con->delay < MAX_DELAY_INTERVAL)
2191                         con->delay *= 2;
2192                 con->ops->get(con);
2193                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2194                                        round_jiffies_relative(con->delay))) {
2195                         dout("fault queued %p delay %lu\n", con, con->delay);
2196                 } else {
2197                         con->ops->put(con);
2198                         dout("fault failed to queue %p delay %lu, backoff\n",
2199                              con, con->delay);
2200                         /*
2201                          * In many cases we see a socket state change
2202                          * while con_work is running and end up
2203                          * queuing (non-delayed) work, such that we
2204                          * can't backoff with a delay.  Set a flag so
2205                          * that when con_work restarts we schedule the
2206                          * delay then.
2207                          */
2208                         set_bit(BACKOFF, &con->state);
2209                 }
2210         }
2211
2212 out_unlock:
2213         mutex_unlock(&con->mutex);
2214 out:
2215         /*
2216          * in case we faulted due to authentication, invalidate our
2217          * current tickets so that we can get new ones.
2218          */
2219         if (con->auth_retry && con->ops->invalidate_authorizer) {
2220                 dout("calling invalidate_authorizer()\n");
2221                 con->ops->invalidate_authorizer(con);
2222         }
2223
2224         if (con->ops->fault)
2225                 con->ops->fault(con);
2226 }
2227
2228
2229
2230 /*
2231  * create a new messenger instance
2232  */
2233 struct ceph_messenger *ceph_messenger_create(struct ceph_entity_addr *myaddr,
2234                                              u32 supported_features,
2235                                              u32 required_features)
2236 {
2237         struct ceph_messenger *msgr;
2238
2239         msgr = kzalloc(sizeof(*msgr), GFP_KERNEL);
2240         if (msgr == NULL)
2241                 return ERR_PTR(-ENOMEM);
2242
2243         msgr->supported_features = supported_features;
2244         msgr->required_features = required_features;
2245
2246         spin_lock_init(&msgr->global_seq_lock);
2247
2248         if (myaddr)
2249                 msgr->inst.addr = *myaddr;
2250
2251         /* select a random nonce */
2252         msgr->inst.addr.type = 0;
2253         get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2254         encode_my_addr(msgr);
2255
2256         dout("messenger_create %p\n", msgr);
2257         return msgr;
2258 }
2259 EXPORT_SYMBOL(ceph_messenger_create);
2260
2261 void ceph_messenger_destroy(struct ceph_messenger *msgr)
2262 {
2263         dout("destroy %p\n", msgr);
2264         kfree(msgr);
2265         dout("destroyed messenger %p\n", msgr);
2266 }
2267 EXPORT_SYMBOL(ceph_messenger_destroy);
2268
2269 static void clear_standby(struct ceph_connection *con)
2270 {
2271         /* come back from STANDBY? */
2272         if (test_and_clear_bit(STANDBY, &con->state)) {
2273                 mutex_lock(&con->mutex);
2274                 dout("clear_standby %p and ++connect_seq\n", con);
2275                 con->connect_seq++;
2276                 WARN_ON(test_bit(WRITE_PENDING, &con->state));
2277                 WARN_ON(test_bit(KEEPALIVE_PENDING, &con->state));
2278                 mutex_unlock(&con->mutex);
2279         }
2280 }
2281
2282 /*
2283  * Queue up an outgoing message on the given connection.
2284  */
2285 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
2286 {
2287         if (test_bit(CLOSED, &con->state)) {
2288                 dout("con_send %p closed, dropping %p\n", con, msg);
2289                 ceph_msg_put(msg);
2290                 return;
2291         }
2292
2293         /* set src+dst */
2294         msg->hdr.src = con->msgr->inst.name;
2295
2296         BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2297
2298         msg->needs_out_seq = true;
2299
2300         /* queue */
2301         mutex_lock(&con->mutex);
2302         BUG_ON(!list_empty(&msg->list_head));
2303         list_add_tail(&msg->list_head, &con->out_queue);
2304         dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
2305              ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
2306              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2307              le32_to_cpu(msg->hdr.front_len),
2308              le32_to_cpu(msg->hdr.middle_len),
2309              le32_to_cpu(msg->hdr.data_len));
2310         mutex_unlock(&con->mutex);
2311
2312         /* if there wasn't anything waiting to send before, queue
2313          * new work */
2314         clear_standby(con);
2315         if (test_and_set_bit(WRITE_PENDING, &con->state) == 0)
2316                 queue_con(con);
2317 }
2318 EXPORT_SYMBOL(ceph_con_send);
2319
2320 /*
2321  * Revoke a message that was previously queued for send
2322  */
2323 void ceph_con_revoke(struct ceph_connection *con, struct ceph_msg *msg)
2324 {
2325         mutex_lock(&con->mutex);
2326         if (!list_empty(&msg->list_head)) {
2327                 dout("con_revoke %p msg %p - was on queue\n", con, msg);
2328                 list_del_init(&msg->list_head);
2329                 ceph_msg_put(msg);
2330                 msg->hdr.seq = 0;
2331         }
2332         if (con->out_msg == msg) {
2333                 dout("con_revoke %p msg %p - was sending\n", con, msg);
2334                 con->out_msg = NULL;
2335                 if (con->out_kvec_is_msg) {
2336                         con->out_skip = con->out_kvec_bytes;
2337                         con->out_kvec_is_msg = false;
2338                 }
2339                 ceph_msg_put(msg);
2340                 msg->hdr.seq = 0;
2341         }
2342         mutex_unlock(&con->mutex);
2343 }
2344
2345 /*
2346  * Revoke a message that we may be reading data into
2347  */
2348 void ceph_con_revoke_message(struct ceph_connection *con, struct ceph_msg *msg)
2349 {
2350         mutex_lock(&con->mutex);
2351         if (con->in_msg && con->in_msg == msg) {
2352                 unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
2353                 unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
2354                 unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
2355
2356                 /* skip rest of message */
2357                 dout("con_revoke_pages %p msg %p revoked\n", con, msg);
2358                         con->in_base_pos = con->in_base_pos -
2359                                 sizeof(struct ceph_msg_header) -
2360                                 front_len -
2361                                 middle_len -
2362                                 data_len -
2363                                 sizeof(struct ceph_msg_footer);
2364                 ceph_msg_put(con->in_msg);
2365                 con->in_msg = NULL;
2366                 con->in_tag = CEPH_MSGR_TAG_READY;
2367                 con->in_seq++;
2368         } else {
2369                 dout("con_revoke_pages %p msg %p pages %p no-op\n",
2370                      con, con->in_msg, msg);
2371         }
2372         mutex_unlock(&con->mutex);
2373 }
2374
2375 /*
2376  * Queue a keepalive byte to ensure the tcp connection is alive.
2377  */
2378 void ceph_con_keepalive(struct ceph_connection *con)
2379 {
2380         dout("con_keepalive %p\n", con);
2381         clear_standby(con);
2382         if (test_and_set_bit(KEEPALIVE_PENDING, &con->state) == 0 &&
2383             test_and_set_bit(WRITE_PENDING, &con->state) == 0)
2384                 queue_con(con);
2385 }
2386 EXPORT_SYMBOL(ceph_con_keepalive);
2387
2388
2389 /*
2390  * construct a new message with given type, size
2391  * the new msg has a ref count of 1.
2392  */
2393 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
2394                               bool can_fail)
2395 {
2396         struct ceph_msg *m;
2397
2398         m = kmalloc(sizeof(*m), flags);
2399         if (m == NULL)
2400                 goto out;
2401         kref_init(&m->kref);
2402         INIT_LIST_HEAD(&m->list_head);
2403
2404         m->hdr.tid = 0;
2405         m->hdr.type = cpu_to_le16(type);
2406         m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
2407         m->hdr.version = 0;
2408         m->hdr.front_len = cpu_to_le32(front_len);
2409         m->hdr.middle_len = 0;
2410         m->hdr.data_len = 0;
2411         m->hdr.data_off = 0;
2412         m->hdr.reserved = 0;
2413         m->footer.front_crc = 0;
2414         m->footer.middle_crc = 0;
2415         m->footer.data_crc = 0;
2416         m->footer.flags = 0;
2417         m->front_max = front_len;
2418         m->front_is_vmalloc = false;
2419         m->more_to_follow = false;
2420         m->ack_stamp = 0;
2421         m->pool = NULL;
2422
2423         /* middle */
2424         m->middle = NULL;
2425
2426         /* data */
2427         m->nr_pages = 0;
2428         m->page_alignment = 0;
2429         m->pages = NULL;
2430         m->pagelist = NULL;
2431         m->bio = NULL;
2432         m->bio_iter = NULL;
2433         m->bio_seg = 0;
2434         m->trail = NULL;
2435
2436         /* front */
2437         if (front_len) {
2438                 if (front_len > PAGE_CACHE_SIZE) {
2439                         m->front.iov_base = __vmalloc(front_len, flags,
2440                                                       PAGE_KERNEL);
2441                         m->front_is_vmalloc = true;
2442                 } else {
2443                         m->front.iov_base = kmalloc(front_len, flags);
2444                 }
2445                 if (m->front.iov_base == NULL) {
2446                         dout("ceph_msg_new can't allocate %d bytes\n",
2447                              front_len);
2448                         goto out2;
2449                 }
2450         } else {
2451                 m->front.iov_base = NULL;
2452         }
2453         m->front.iov_len = front_len;
2454
2455         dout("ceph_msg_new %p front %d\n", m, front_len);
2456         return m;
2457
2458 out2:
2459         ceph_msg_put(m);
2460 out:
2461         if (!can_fail) {
2462                 pr_err("msg_new can't create type %d front %d\n", type,
2463                        front_len);
2464                 WARN_ON(1);
2465         } else {
2466                 dout("msg_new can't create type %d front %d\n", type,
2467                      front_len);
2468         }
2469         return NULL;
2470 }
2471 EXPORT_SYMBOL(ceph_msg_new);
2472
2473 /*
2474  * Allocate "middle" portion of a message, if it is needed and wasn't
2475  * allocated by alloc_msg.  This allows us to read a small fixed-size
2476  * per-type header in the front and then gracefully fail (i.e.,
2477  * propagate the error to the caller based on info in the front) when
2478  * the middle is too large.
2479  */
2480 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
2481 {
2482         int type = le16_to_cpu(msg->hdr.type);
2483         int middle_len = le32_to_cpu(msg->hdr.middle_len);
2484
2485         dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
2486              ceph_msg_type_name(type), middle_len);
2487         BUG_ON(!middle_len);
2488         BUG_ON(msg->middle);
2489
2490         msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
2491         if (!msg->middle)
2492                 return -ENOMEM;
2493         return 0;
2494 }
2495
2496 /*
2497  * Generic message allocator, for incoming messages.
2498  */
2499 static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
2500                                 struct ceph_msg_header *hdr,
2501                                 int *skip)
2502 {
2503         int type = le16_to_cpu(hdr->type);
2504         int front_len = le32_to_cpu(hdr->front_len);
2505         int middle_len = le32_to_cpu(hdr->middle_len);
2506         struct ceph_msg *msg = NULL;
2507         int ret;
2508
2509         if (con->ops->alloc_msg) {
2510                 mutex_unlock(&con->mutex);
2511                 msg = con->ops->alloc_msg(con, hdr, skip);
2512                 mutex_lock(&con->mutex);
2513                 if (!msg || *skip)
2514                         return NULL;
2515         }
2516         if (!msg) {
2517                 *skip = 0;
2518                 msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
2519                 if (!msg) {
2520                         pr_err("unable to allocate msg type %d len %d\n",
2521                                type, front_len);
2522                         return NULL;
2523                 }
2524                 msg->page_alignment = le16_to_cpu(hdr->data_off);
2525         }
2526         memcpy(&msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2527
2528         if (middle_len && !msg->middle) {
2529                 ret = ceph_alloc_middle(con, msg);
2530                 if (ret < 0) {
2531                         ceph_msg_put(msg);
2532                         return NULL;
2533                 }
2534         }
2535
2536         return msg;
2537 }
2538
2539
2540 /*
2541  * Free a generically kmalloc'd message.
2542  */
2543 void ceph_msg_kfree(struct ceph_msg *m)
2544 {
2545         dout("msg_kfree %p\n", m);
2546         if (m->front_is_vmalloc)
2547                 vfree(m->front.iov_base);
2548         else
2549                 kfree(m->front.iov_base);
2550         kfree(m);
2551 }
2552
2553 /*
2554  * Drop a msg ref.  Destroy as needed.
2555  */
2556 void ceph_msg_last_put(struct kref *kref)
2557 {
2558         struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
2559
2560         dout("ceph_msg_put last one on %p\n", m);
2561         WARN_ON(!list_empty(&m->list_head));
2562
2563         /* drop middle, data, if any */
2564         if (m->middle) {
2565                 ceph_buffer_put(m->middle);
2566                 m->middle = NULL;
2567         }
2568         m->nr_pages = 0;
2569         m->pages = NULL;
2570
2571         if (m->pagelist) {
2572                 ceph_pagelist_release(m->pagelist);
2573                 kfree(m->pagelist);
2574                 m->pagelist = NULL;
2575         }
2576
2577         m->trail = NULL;
2578
2579         if (m->pool)
2580                 ceph_msgpool_put(m->pool, m);
2581         else
2582                 ceph_msg_kfree(m);
2583 }
2584 EXPORT_SYMBOL(ceph_msg_last_put);
2585
2586 void ceph_msg_dump(struct ceph_msg *msg)
2587 {
2588         pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
2589                  msg->front_max, msg->nr_pages);
2590         print_hex_dump(KERN_DEBUG, "header: ",
2591                        DUMP_PREFIX_OFFSET, 16, 1,
2592                        &msg->hdr, sizeof(msg->hdr), true);
2593         print_hex_dump(KERN_DEBUG, " front: ",
2594                        DUMP_PREFIX_OFFSET, 16, 1,
2595                        msg->front.iov_base, msg->front.iov_len, true);
2596         if (msg->middle)
2597                 print_hex_dump(KERN_DEBUG, "middle: ",
2598                                DUMP_PREFIX_OFFSET, 16, 1,
2599                                msg->middle->vec.iov_base,
2600                                msg->middle->vec.iov_len, true);
2601         print_hex_dump(KERN_DEBUG, "footer: ",
2602                        DUMP_PREFIX_OFFSET, 16, 1,
2603                        &msg->footer, sizeof(msg->footer), true);
2604 }
2605 EXPORT_SYMBOL(ceph_msg_dump);