~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/ceph/messenger.c

Version: ~ [ linux-5.5-rc1 ] ~ [ linux-5.4.2 ] ~ [ linux-5.3.15 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.88 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.158 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.206 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.206 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.78 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp