messenger.h 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381
  1. /* SPDX-License-Identifier: GPL-2.0 */
  2. #ifndef __FS_CEPH_MESSENGER_H
  3. #define __FS_CEPH_MESSENGER_H
  4. #include <linux/bvec.h>
  5. #include <linux/kref.h>
  6. #include <linux/mutex.h>
  7. #include <linux/net.h>
  8. #include <linux/radix-tree.h>
  9. #include <linux/uio.h>
  10. #include <linux/workqueue.h>
  11. #include <net/net_namespace.h>
  12. #include <linux/ceph/types.h>
  13. #include <linux/ceph/buffer.h>
  14. struct ceph_msg;
  15. struct ceph_connection;
  16. /*
  17. * Ceph defines these callbacks for handling connection events.
  18. */
  19. struct ceph_connection_operations {
  20. struct ceph_connection *(*get)(struct ceph_connection *);
  21. void (*put)(struct ceph_connection *);
  22. /* handle an incoming message. */
  23. void (*dispatch) (struct ceph_connection *con, struct ceph_msg *m);
  24. /* authorize an outgoing connection */
  25. struct ceph_auth_handshake *(*get_authorizer) (
  26. struct ceph_connection *con,
  27. int *proto, int force_new);
  28. int (*add_authorizer_challenge)(struct ceph_connection *con,
  29. void *challenge_buf,
  30. int challenge_buf_len);
  31. int (*verify_authorizer_reply) (struct ceph_connection *con);
  32. int (*invalidate_authorizer)(struct ceph_connection *con);
  33. /* there was some error on the socket (disconnect, whatever) */
  34. void (*fault) (struct ceph_connection *con);
  35. /* a remote host as terminated a message exchange session, and messages
  36. * we sent (or they tried to send us) may be lost. */
  37. void (*peer_reset) (struct ceph_connection *con);
  38. struct ceph_msg * (*alloc_msg) (struct ceph_connection *con,
  39. struct ceph_msg_header *hdr,
  40. int *skip);
  41. void (*reencode_message) (struct ceph_msg *msg);
  42. int (*sign_message) (struct ceph_msg *msg);
  43. int (*check_message_signature) (struct ceph_msg *msg);
  44. };
  45. /* use format string %s%lld */
  46. #define ENTITY_NAME(n) ceph_entity_type_name((n).type), le64_to_cpu((n).num)
  47. struct ceph_messenger {
  48. struct ceph_entity_inst inst; /* my name+address */
  49. struct ceph_entity_addr my_enc_addr;
  50. atomic_t stopping;
  51. possible_net_t net;
  52. /*
  53. * the global_seq counts connections i (attempt to) initiate
  54. * in order to disambiguate certain connect race conditions.
  55. */
  56. u32 global_seq;
  57. spinlock_t global_seq_lock;
  58. };
  59. enum ceph_msg_data_type {
  60. CEPH_MSG_DATA_NONE, /* message contains no data payload */
  61. CEPH_MSG_DATA_PAGES, /* data source/destination is a page array */
  62. CEPH_MSG_DATA_PAGELIST, /* data source/destination is a pagelist */
  63. #ifdef CONFIG_BLOCK
  64. CEPH_MSG_DATA_BIO, /* data source/destination is a bio list */
  65. #endif /* CONFIG_BLOCK */
  66. CEPH_MSG_DATA_BVECS, /* data source/destination is a bio_vec array */
  67. };
  68. #ifdef CONFIG_BLOCK
  69. struct ceph_bio_iter {
  70. struct bio *bio;
  71. struct bvec_iter iter;
  72. };
  73. #define __ceph_bio_iter_advance_step(it, n, STEP) do { \
  74. unsigned int __n = (n), __cur_n; \
  75. \
  76. while (__n) { \
  77. BUG_ON(!(it)->iter.bi_size); \
  78. __cur_n = min((it)->iter.bi_size, __n); \
  79. (void)(STEP); \
  80. bio_advance_iter((it)->bio, &(it)->iter, __cur_n); \
  81. if (!(it)->iter.bi_size && (it)->bio->bi_next) { \
  82. dout("__ceph_bio_iter_advance_step next bio\n"); \
  83. (it)->bio = (it)->bio->bi_next; \
  84. (it)->iter = (it)->bio->bi_iter; \
  85. } \
  86. __n -= __cur_n; \
  87. } \
  88. } while (0)
  89. /*
  90. * Advance @it by @n bytes.
  91. */
  92. #define ceph_bio_iter_advance(it, n) \
  93. __ceph_bio_iter_advance_step(it, n, 0)
  94. /*
  95. * Advance @it by @n bytes, executing BVEC_STEP for each bio_vec.
  96. */
  97. #define ceph_bio_iter_advance_step(it, n, BVEC_STEP) \
  98. __ceph_bio_iter_advance_step(it, n, ({ \
  99. struct bio_vec bv; \
  100. struct bvec_iter __cur_iter; \
  101. \
  102. __cur_iter = (it)->iter; \
  103. __cur_iter.bi_size = __cur_n; \
  104. __bio_for_each_segment(bv, (it)->bio, __cur_iter, __cur_iter) \
  105. (void)(BVEC_STEP); \
  106. }))
  107. #endif /* CONFIG_BLOCK */
  108. struct ceph_bvec_iter {
  109. struct bio_vec *bvecs;
  110. struct bvec_iter iter;
  111. };
  112. #define __ceph_bvec_iter_advance_step(it, n, STEP) do { \
  113. BUG_ON((n) > (it)->iter.bi_size); \
  114. (void)(STEP); \
  115. bvec_iter_advance((it)->bvecs, &(it)->iter, (n)); \
  116. } while (0)
  117. /*
  118. * Advance @it by @n bytes.
  119. */
  120. #define ceph_bvec_iter_advance(it, n) \
  121. __ceph_bvec_iter_advance_step(it, n, 0)
  122. /*
  123. * Advance @it by @n bytes, executing BVEC_STEP for each bio_vec.
  124. */
  125. #define ceph_bvec_iter_advance_step(it, n, BVEC_STEP) \
  126. __ceph_bvec_iter_advance_step(it, n, ({ \
  127. struct bio_vec bv; \
  128. struct bvec_iter __cur_iter; \
  129. \
  130. __cur_iter = (it)->iter; \
  131. __cur_iter.bi_size = (n); \
  132. for_each_bvec(bv, (it)->bvecs, __cur_iter, __cur_iter) \
  133. (void)(BVEC_STEP); \
  134. }))
  135. #define ceph_bvec_iter_shorten(it, n) do { \
  136. BUG_ON((n) > (it)->iter.bi_size); \
  137. (it)->iter.bi_size = (n); \
  138. } while (0)
  139. struct ceph_msg_data {
  140. enum ceph_msg_data_type type;
  141. union {
  142. #ifdef CONFIG_BLOCK
  143. struct {
  144. struct ceph_bio_iter bio_pos;
  145. u32 bio_length;
  146. };
  147. #endif /* CONFIG_BLOCK */
  148. struct ceph_bvec_iter bvec_pos;
  149. struct {
  150. struct page **pages;
  151. size_t length; /* total # bytes */
  152. unsigned int alignment; /* first page */
  153. bool own_pages;
  154. };
  155. struct ceph_pagelist *pagelist;
  156. };
  157. };
  158. struct ceph_msg_data_cursor {
  159. size_t total_resid; /* across all data items */
  160. struct ceph_msg_data *data; /* current data item */
  161. size_t resid; /* bytes not yet consumed */
  162. bool last_piece; /* current is last piece */
  163. bool need_crc; /* crc update needed */
  164. union {
  165. #ifdef CONFIG_BLOCK
  166. struct ceph_bio_iter bio_iter;
  167. #endif /* CONFIG_BLOCK */
  168. struct bvec_iter bvec_iter;
  169. struct { /* pages */
  170. unsigned int page_offset; /* offset in page */
  171. unsigned short page_index; /* index in array */
  172. unsigned short page_count; /* pages in array */
  173. };
  174. struct { /* pagelist */
  175. struct page *page; /* page from list */
  176. size_t offset; /* bytes from list */
  177. };
  178. };
  179. };
  180. /*
  181. * a single message. it contains a header (src, dest, message type, etc.),
  182. * footer (crc values, mainly), a "front" message body, and possibly a
  183. * data payload (stored in some number of pages).
  184. */
  185. struct ceph_msg {
  186. struct ceph_msg_header hdr; /* header */
  187. union {
  188. struct ceph_msg_footer footer; /* footer */
  189. struct ceph_msg_footer_old old_footer; /* old format footer */
  190. };
  191. struct kvec front; /* unaligned blobs of message */
  192. struct ceph_buffer *middle;
  193. size_t data_length;
  194. struct ceph_msg_data *data;
  195. int num_data_items;
  196. int max_data_items;
  197. struct ceph_msg_data_cursor cursor;
  198. struct ceph_connection *con;
  199. struct list_head list_head; /* links for connection lists */
  200. struct kref kref;
  201. bool more_to_follow;
  202. bool needs_out_seq;
  203. int front_alloc_len;
  204. unsigned long ack_stamp; /* tx: when we were acked */
  205. struct ceph_msgpool *pool;
  206. };
  207. /* ceph connection fault delay defaults, for exponential backoff */
  208. #define BASE_DELAY_INTERVAL (HZ/2)
  209. #define MAX_DELAY_INTERVAL (5 * 60 * HZ)
  210. /*
  211. * A single connection with another host.
  212. *
  213. * We maintain a queue of outgoing messages, and some session state to
  214. * ensure that we can preserve the lossless, ordered delivery of
  215. * messages in the case of a TCP disconnect.
  216. */
  217. struct ceph_connection {
  218. void *private;
  219. const struct ceph_connection_operations *ops;
  220. struct ceph_messenger *msgr;
  221. atomic_t sock_state;
  222. struct socket *sock;
  223. struct ceph_entity_addr peer_addr; /* peer address */
  224. struct ceph_entity_addr peer_addr_for_me;
  225. unsigned long flags;
  226. unsigned long state;
  227. const char *error_msg; /* error message, if any */
  228. struct ceph_entity_name peer_name; /* peer name */
  229. u64 peer_features;
  230. u32 connect_seq; /* identify the most recent connection
  231. attempt for this connection, client */
  232. u32 peer_global_seq; /* peer's global seq for this connection */
  233. struct ceph_auth_handshake *auth;
  234. int auth_retry; /* true if we need a newer authorizer */
  235. struct mutex mutex;
  236. /* out queue */
  237. struct list_head out_queue;
  238. struct list_head out_sent; /* sending or sent but unacked */
  239. u64 out_seq; /* last message queued for send */
  240. u64 in_seq, in_seq_acked; /* last message received, acked */
  241. /* connection negotiation temps */
  242. char in_banner[CEPH_BANNER_MAX_LEN];
  243. struct ceph_msg_connect out_connect;
  244. struct ceph_msg_connect_reply in_reply;
  245. struct ceph_entity_addr actual_peer_addr;
  246. /* message out temps */
  247. struct ceph_msg_header out_hdr;
  248. struct ceph_msg *out_msg; /* sending message (== tail of
  249. out_sent) */
  250. bool out_msg_done;
  251. struct kvec out_kvec[8], /* sending header/footer data */
  252. *out_kvec_cur;
  253. int out_kvec_left; /* kvec's left in out_kvec */
  254. int out_skip; /* skip this many bytes */
  255. int out_kvec_bytes; /* total bytes left */
  256. int out_more; /* there is more data after the kvecs */
  257. __le64 out_temp_ack; /* for writing an ack */
  258. struct ceph_timespec out_temp_keepalive2; /* for writing keepalive2
  259. stamp */
  260. /* message in temps */
  261. struct ceph_msg_header in_hdr;
  262. struct ceph_msg *in_msg;
  263. u32 in_front_crc, in_middle_crc, in_data_crc; /* calculated crc */
  264. char in_tag; /* protocol control byte */
  265. int in_base_pos; /* bytes read */
  266. __le64 in_temp_ack; /* for reading an ack */
  267. struct timespec64 last_keepalive_ack; /* keepalive2 ack stamp */
  268. struct delayed_work work; /* send|recv work */
  269. unsigned long delay; /* current delay interval */
  270. };
  271. extern const char *ceph_pr_addr(const struct ceph_entity_addr *addr);
  272. extern int ceph_parse_ips(const char *c, const char *end,
  273. struct ceph_entity_addr *addr,
  274. int max_count, int *count);
  275. extern int ceph_msgr_init(void);
  276. extern void ceph_msgr_exit(void);
  277. extern void ceph_msgr_flush(void);
  278. extern void ceph_messenger_init(struct ceph_messenger *msgr,
  279. struct ceph_entity_addr *myaddr);
  280. extern void ceph_messenger_fini(struct ceph_messenger *msgr);
  281. extern void ceph_messenger_reset_nonce(struct ceph_messenger *msgr);
  282. extern void ceph_con_init(struct ceph_connection *con, void *private,
  283. const struct ceph_connection_operations *ops,
  284. struct ceph_messenger *msgr);
  285. extern void ceph_con_open(struct ceph_connection *con,
  286. __u8 entity_type, __u64 entity_num,
  287. struct ceph_entity_addr *addr);
  288. extern bool ceph_con_opened(struct ceph_connection *con);
  289. extern void ceph_con_close(struct ceph_connection *con);
  290. extern void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg);
  291. extern void ceph_msg_revoke(struct ceph_msg *msg);
  292. extern void ceph_msg_revoke_incoming(struct ceph_msg *msg);
  293. extern void ceph_con_keepalive(struct ceph_connection *con);
  294. extern bool ceph_con_keepalive_expired(struct ceph_connection *con,
  295. unsigned long interval);
  296. void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
  297. size_t length, size_t alignment, bool own_pages);
  298. extern void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
  299. struct ceph_pagelist *pagelist);
  300. #ifdef CONFIG_BLOCK
  301. void ceph_msg_data_add_bio(struct ceph_msg *msg, struct ceph_bio_iter *bio_pos,
  302. u32 length);
  303. #endif /* CONFIG_BLOCK */
  304. void ceph_msg_data_add_bvecs(struct ceph_msg *msg,
  305. struct ceph_bvec_iter *bvec_pos);
  306. struct ceph_msg *ceph_msg_new2(int type, int front_len, int max_data_items,
  307. gfp_t flags, bool can_fail);
  308. extern struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
  309. bool can_fail);
  310. extern struct ceph_msg *ceph_msg_get(struct ceph_msg *msg);
  311. extern void ceph_msg_put(struct ceph_msg *msg);
  312. extern void ceph_msg_dump(struct ceph_msg *msg);
  313. #endif