recv.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831
  1. /*
  2. * Copyright (c) 2006, 2019 Oracle and/or its affiliates. All rights reserved.
  3. *
  4. * This software is available to you under a choice of one of two
  5. * licenses. You may choose to be licensed under the terms of the GNU
  6. * General Public License (GPL) Version 2, available from the file
  7. * COPYING in the main directory of this source tree, or the
  8. * OpenIB.org BSD license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or
  11. * without modification, are permitted provided that the following
  12. * conditions are met:
  13. *
  14. * - Redistributions of source code must retain the above
  15. * copyright notice, this list of conditions and the following
  16. * disclaimer.
  17. *
  18. * - Redistributions in binary form must reproduce the above
  19. * copyright notice, this list of conditions and the following
  20. * disclaimer in the documentation and/or other materials
  21. * provided with the distribution.
  22. *
  23. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  24. * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  25. * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  26. * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  27. * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  28. * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  29. * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  30. * SOFTWARE.
  31. *
  32. */
  33. #include <linux/kernel.h>
  34. #include <linux/slab.h>
  35. #include <net/sock.h>
  36. #include <linux/in.h>
  37. #include <linux/export.h>
  38. #include <linux/time.h>
  39. #include <linux/rds.h>
  40. #include "rds.h"
  41. void rds_inc_init(struct rds_incoming *inc, struct rds_connection *conn,
  42. struct in6_addr *saddr)
  43. {
  44. refcount_set(&inc->i_refcount, 1);
  45. INIT_LIST_HEAD(&inc->i_item);
  46. inc->i_conn = conn;
  47. inc->i_saddr = *saddr;
  48. inc->i_usercopy.rdma_cookie = 0;
  49. inc->i_usercopy.rx_tstamp = ktime_set(0, 0);
  50. memset(inc->i_rx_lat_trace, 0, sizeof(inc->i_rx_lat_trace));
  51. }
  52. EXPORT_SYMBOL_GPL(rds_inc_init);
  53. void rds_inc_path_init(struct rds_incoming *inc, struct rds_conn_path *cp,
  54. struct in6_addr *saddr)
  55. {
  56. refcount_set(&inc->i_refcount, 1);
  57. INIT_LIST_HEAD(&inc->i_item);
  58. inc->i_conn = cp->cp_conn;
  59. inc->i_conn_path = cp;
  60. inc->i_saddr = *saddr;
  61. inc->i_usercopy.rdma_cookie = 0;
  62. inc->i_usercopy.rx_tstamp = ktime_set(0, 0);
  63. }
  64. EXPORT_SYMBOL_GPL(rds_inc_path_init);
  65. static void rds_inc_addref(struct rds_incoming *inc)
  66. {
  67. rdsdebug("addref inc %p ref %d\n", inc, refcount_read(&inc->i_refcount));
  68. refcount_inc(&inc->i_refcount);
  69. }
  70. void rds_inc_put(struct rds_incoming *inc)
  71. {
  72. rdsdebug("put inc %p ref %d\n", inc, refcount_read(&inc->i_refcount));
  73. if (refcount_dec_and_test(&inc->i_refcount)) {
  74. BUG_ON(!list_empty(&inc->i_item));
  75. inc->i_conn->c_trans->inc_free(inc);
  76. }
  77. }
  78. EXPORT_SYMBOL_GPL(rds_inc_put);
  79. static void rds_recv_rcvbuf_delta(struct rds_sock *rs, struct sock *sk,
  80. struct rds_cong_map *map,
  81. int delta, __be16 port)
  82. {
  83. int now_congested;
  84. if (delta == 0)
  85. return;
  86. rs->rs_rcv_bytes += delta;
  87. if (delta > 0)
  88. rds_stats_add(s_recv_bytes_added_to_socket, delta);
  89. else
  90. rds_stats_add(s_recv_bytes_removed_from_socket, -delta);
  91. /* loop transport doesn't send/recv congestion updates */
  92. if (rs->rs_transport->t_type == RDS_TRANS_LOOP)
  93. return;
  94. now_congested = rs->rs_rcv_bytes > rds_sk_rcvbuf(rs);
  95. rdsdebug("rs %p (%pI6c:%u) recv bytes %d buf %d "
  96. "now_cong %d delta %d\n",
  97. rs, &rs->rs_bound_addr,
  98. ntohs(rs->rs_bound_port), rs->rs_rcv_bytes,
  99. rds_sk_rcvbuf(rs), now_congested, delta);
  100. /* wasn't -> am congested */
  101. if (!rs->rs_congested && now_congested) {
  102. rs->rs_congested = 1;
  103. rds_cong_set_bit(map, port);
  104. rds_cong_queue_updates(map);
  105. }
  106. /* was -> aren't congested */
  107. /* Require more free space before reporting uncongested to prevent
  108. bouncing cong/uncong state too often */
  109. else if (rs->rs_congested && (rs->rs_rcv_bytes < (rds_sk_rcvbuf(rs)/2))) {
  110. rs->rs_congested = 0;
  111. rds_cong_clear_bit(map, port);
  112. rds_cong_queue_updates(map);
  113. }
  114. /* do nothing if no change in cong state */
  115. }
  116. static void rds_conn_peer_gen_update(struct rds_connection *conn,
  117. u32 peer_gen_num)
  118. {
  119. int i;
  120. struct rds_message *rm, *tmp;
  121. unsigned long flags;
  122. WARN_ON(conn->c_trans->t_type != RDS_TRANS_TCP);
  123. if (peer_gen_num != 0) {
  124. if (conn->c_peer_gen_num != 0 &&
  125. peer_gen_num != conn->c_peer_gen_num) {
  126. for (i = 0; i < RDS_MPATH_WORKERS; i++) {
  127. struct rds_conn_path *cp;
  128. cp = &conn->c_path[i];
  129. spin_lock_irqsave(&cp->cp_lock, flags);
  130. cp->cp_next_tx_seq = 1;
  131. cp->cp_next_rx_seq = 0;
  132. list_for_each_entry_safe(rm, tmp,
  133. &cp->cp_retrans,
  134. m_conn_item) {
  135. set_bit(RDS_MSG_FLUSH, &rm->m_flags);
  136. }
  137. spin_unlock_irqrestore(&cp->cp_lock, flags);
  138. }
  139. }
  140. conn->c_peer_gen_num = peer_gen_num;
  141. }
  142. }
  143. /*
  144. * Process all extension headers that come with this message.
  145. */
  146. static void rds_recv_incoming_exthdrs(struct rds_incoming *inc, struct rds_sock *rs)
  147. {
  148. struct rds_header *hdr = &inc->i_hdr;
  149. unsigned int pos = 0, type, len;
  150. union {
  151. struct rds_ext_header_version version;
  152. struct rds_ext_header_rdma rdma;
  153. struct rds_ext_header_rdma_dest rdma_dest;
  154. } buffer;
  155. while (1) {
  156. len = sizeof(buffer);
  157. type = rds_message_next_extension(hdr, &pos, &buffer, &len);
  158. if (type == RDS_EXTHDR_NONE)
  159. break;
  160. /* Process extension header here */
  161. switch (type) {
  162. case RDS_EXTHDR_RDMA:
  163. rds_rdma_unuse(rs, be32_to_cpu(buffer.rdma.h_rdma_rkey), 0);
  164. break;
  165. case RDS_EXTHDR_RDMA_DEST:
  166. /* We ignore the size for now. We could stash it
  167. * somewhere and use it for error checking. */
  168. inc->i_usercopy.rdma_cookie = rds_rdma_make_cookie(
  169. be32_to_cpu(buffer.rdma_dest.h_rdma_rkey),
  170. be32_to_cpu(buffer.rdma_dest.h_rdma_offset));
  171. break;
  172. }
  173. }
  174. }
  175. static void rds_recv_hs_exthdrs(struct rds_header *hdr,
  176. struct rds_connection *conn)
  177. {
  178. unsigned int pos = 0, type, len;
  179. union {
  180. struct rds_ext_header_version version;
  181. u16 rds_npaths;
  182. u32 rds_gen_num;
  183. } buffer;
  184. u32 new_peer_gen_num = 0;
  185. while (1) {
  186. len = sizeof(buffer);
  187. type = rds_message_next_extension(hdr, &pos, &buffer, &len);
  188. if (type == RDS_EXTHDR_NONE)
  189. break;
  190. /* Process extension header here */
  191. switch (type) {
  192. case RDS_EXTHDR_NPATHS:
  193. conn->c_npaths = min_t(int, RDS_MPATH_WORKERS,
  194. be16_to_cpu(buffer.rds_npaths));
  195. break;
  196. case RDS_EXTHDR_GEN_NUM:
  197. new_peer_gen_num = be32_to_cpu(buffer.rds_gen_num);
  198. break;
  199. default:
  200. pr_warn_ratelimited("ignoring unknown exthdr type "
  201. "0x%x\n", type);
  202. }
  203. }
  204. /* if RDS_EXTHDR_NPATHS was not found, default to a single-path */
  205. conn->c_npaths = max_t(int, conn->c_npaths, 1);
  206. conn->c_ping_triggered = 0;
  207. rds_conn_peer_gen_update(conn, new_peer_gen_num);
  208. }
  209. /* rds_start_mprds() will synchronously start multiple paths when appropriate.
  210. * The scheme is based on the following rules:
  211. *
  212. * 1. rds_sendmsg on first connect attempt sends the probe ping, with the
  213. * sender's npaths (s_npaths)
  214. * 2. rcvr of probe-ping knows the mprds_paths = min(s_npaths, r_npaths). It
  215. * sends back a probe-pong with r_npaths. After that, if rcvr is the
  216. * smaller ip addr, it starts rds_conn_path_connect_if_down on all
  217. * mprds_paths.
  218. * 3. sender gets woken up, and can move to rds_conn_path_connect_if_down.
  219. * If it is the smaller ipaddr, rds_conn_path_connect_if_down can be
  220. * called after reception of the probe-pong on all mprds_paths.
  221. * Otherwise (sender of probe-ping is not the smaller ip addr): just call
  222. * rds_conn_path_connect_if_down on the hashed path. (see rule 4)
  223. * 4. rds_connect_worker must only trigger a connection if laddr < faddr.
  224. * 5. sender may end up queuing the packet on the cp. will get sent out later.
  225. * when connection is completed.
  226. */
  227. static void rds_start_mprds(struct rds_connection *conn)
  228. {
  229. int i;
  230. struct rds_conn_path *cp;
  231. if (conn->c_npaths > 1 &&
  232. rds_addr_cmp(&conn->c_laddr, &conn->c_faddr) < 0) {
  233. for (i = 0; i < conn->c_npaths; i++) {
  234. cp = &conn->c_path[i];
  235. rds_conn_path_connect_if_down(cp);
  236. }
  237. }
  238. }
  239. /*
  240. * The transport must make sure that this is serialized against other
  241. * rx and conn reset on this specific conn.
  242. *
  243. * We currently assert that only one fragmented message will be sent
  244. * down a connection at a time. This lets us reassemble in the conn
  245. * instead of per-flow which means that we don't have to go digging through
  246. * flows to tear down partial reassembly progress on conn failure and
  247. * we save flow lookup and locking for each frag arrival. It does mean
  248. * that small messages will wait behind large ones. Fragmenting at all
  249. * is only to reduce the memory consumption of pre-posted buffers.
  250. *
  251. * The caller passes in saddr and daddr instead of us getting it from the
  252. * conn. This lets loopback, who only has one conn for both directions,
  253. * tell us which roles the addrs in the conn are playing for this message.
  254. */
  255. void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,
  256. struct in6_addr *daddr,
  257. struct rds_incoming *inc, gfp_t gfp)
  258. {
  259. struct rds_sock *rs = NULL;
  260. struct sock *sk;
  261. unsigned long flags;
  262. struct rds_conn_path *cp;
  263. inc->i_conn = conn;
  264. inc->i_rx_jiffies = jiffies;
  265. if (conn->c_trans->t_mp_capable)
  266. cp = inc->i_conn_path;
  267. else
  268. cp = &conn->c_path[0];
  269. rdsdebug("conn %p next %llu inc %p seq %llu len %u sport %u dport %u "
  270. "flags 0x%x rx_jiffies %lu\n", conn,
  271. (unsigned long long)cp->cp_next_rx_seq,
  272. inc,
  273. (unsigned long long)be64_to_cpu(inc->i_hdr.h_sequence),
  274. be32_to_cpu(inc->i_hdr.h_len),
  275. be16_to_cpu(inc->i_hdr.h_sport),
  276. be16_to_cpu(inc->i_hdr.h_dport),
  277. inc->i_hdr.h_flags,
  278. inc->i_rx_jiffies);
  279. /*
  280. * Sequence numbers should only increase. Messages get their
  281. * sequence number as they're queued in a sending conn. They
  282. * can be dropped, though, if the sending socket is closed before
  283. * they hit the wire. So sequence numbers can skip forward
  284. * under normal operation. They can also drop back in the conn
  285. * failover case as previously sent messages are resent down the
  286. * new instance of a conn. We drop those, otherwise we have
  287. * to assume that the next valid seq does not come after a
  288. * hole in the fragment stream.
  289. *
  290. * The headers don't give us a way to realize if fragments of
  291. * a message have been dropped. We assume that frags that arrive
  292. * to a flow are part of the current message on the flow that is
  293. * being reassembled. This means that senders can't drop messages
  294. * from the sending conn until all their frags are sent.
  295. *
  296. * XXX we could spend more on the wire to get more robust failure
  297. * detection, arguably worth it to avoid data corruption.
  298. */
  299. if (be64_to_cpu(inc->i_hdr.h_sequence) < cp->cp_next_rx_seq &&
  300. (inc->i_hdr.h_flags & RDS_FLAG_RETRANSMITTED)) {
  301. rds_stats_inc(s_recv_drop_old_seq);
  302. goto out;
  303. }
  304. cp->cp_next_rx_seq = be64_to_cpu(inc->i_hdr.h_sequence) + 1;
  305. if (rds_sysctl_ping_enable && inc->i_hdr.h_dport == 0) {
  306. if (inc->i_hdr.h_sport == 0) {
  307. rdsdebug("ignore ping with 0 sport from %pI6c\n",
  308. saddr);
  309. goto out;
  310. }
  311. rds_stats_inc(s_recv_ping);
  312. rds_send_pong(cp, inc->i_hdr.h_sport);
  313. /* if this is a handshake ping, start multipath if necessary */
  314. if (RDS_HS_PROBE(be16_to_cpu(inc->i_hdr.h_sport),
  315. be16_to_cpu(inc->i_hdr.h_dport))) {
  316. rds_recv_hs_exthdrs(&inc->i_hdr, cp->cp_conn);
  317. rds_start_mprds(cp->cp_conn);
  318. }
  319. goto out;
  320. }
  321. if (be16_to_cpu(inc->i_hdr.h_dport) == RDS_FLAG_PROBE_PORT &&
  322. inc->i_hdr.h_sport == 0) {
  323. rds_recv_hs_exthdrs(&inc->i_hdr, cp->cp_conn);
  324. /* if this is a handshake pong, start multipath if necessary */
  325. rds_start_mprds(cp->cp_conn);
  326. wake_up(&cp->cp_conn->c_hs_waitq);
  327. goto out;
  328. }
  329. rs = rds_find_bound(daddr, inc->i_hdr.h_dport, conn->c_bound_if);
  330. if (!rs) {
  331. rds_stats_inc(s_recv_drop_no_sock);
  332. goto out;
  333. }
  334. /* Process extension headers */
  335. rds_recv_incoming_exthdrs(inc, rs);
  336. /* We can be racing with rds_release() which marks the socket dead. */
  337. sk = rds_rs_to_sk(rs);
  338. /* serialize with rds_release -> sock_orphan */
  339. write_lock_irqsave(&rs->rs_recv_lock, flags);
  340. if (!sock_flag(sk, SOCK_DEAD)) {
  341. rdsdebug("adding inc %p to rs %p's recv queue\n", inc, rs);
  342. rds_stats_inc(s_recv_queued);
  343. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  344. be32_to_cpu(inc->i_hdr.h_len),
  345. inc->i_hdr.h_dport);
  346. if (sock_flag(sk, SOCK_RCVTSTAMP))
  347. inc->i_usercopy.rx_tstamp = ktime_get_real();
  348. rds_inc_addref(inc);
  349. inc->i_rx_lat_trace[RDS_MSG_RX_END] = local_clock();
  350. list_add_tail(&inc->i_item, &rs->rs_recv_queue);
  351. __rds_wake_sk_sleep(sk);
  352. } else {
  353. rds_stats_inc(s_recv_drop_dead_sock);
  354. }
  355. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  356. out:
  357. if (rs)
  358. rds_sock_put(rs);
  359. }
  360. EXPORT_SYMBOL_GPL(rds_recv_incoming);
  361. /*
  362. * be very careful here. This is being called as the condition in
  363. * wait_event_*() needs to cope with being called many times.
  364. */
  365. static int rds_next_incoming(struct rds_sock *rs, struct rds_incoming **inc)
  366. {
  367. unsigned long flags;
  368. if (!*inc) {
  369. read_lock_irqsave(&rs->rs_recv_lock, flags);
  370. if (!list_empty(&rs->rs_recv_queue)) {
  371. *inc = list_entry(rs->rs_recv_queue.next,
  372. struct rds_incoming,
  373. i_item);
  374. rds_inc_addref(*inc);
  375. }
  376. read_unlock_irqrestore(&rs->rs_recv_lock, flags);
  377. }
  378. return *inc != NULL;
  379. }
  380. static int rds_still_queued(struct rds_sock *rs, struct rds_incoming *inc,
  381. int drop)
  382. {
  383. struct sock *sk = rds_rs_to_sk(rs);
  384. int ret = 0;
  385. unsigned long flags;
  386. write_lock_irqsave(&rs->rs_recv_lock, flags);
  387. if (!list_empty(&inc->i_item)) {
  388. ret = 1;
  389. if (drop) {
  390. /* XXX make sure this i_conn is reliable */
  391. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  392. -be32_to_cpu(inc->i_hdr.h_len),
  393. inc->i_hdr.h_dport);
  394. list_del_init(&inc->i_item);
  395. rds_inc_put(inc);
  396. }
  397. }
  398. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  399. rdsdebug("inc %p rs %p still %d dropped %d\n", inc, rs, ret, drop);
  400. return ret;
  401. }
  402. /*
  403. * Pull errors off the error queue.
  404. * If msghdr is NULL, we will just purge the error queue.
  405. */
  406. int rds_notify_queue_get(struct rds_sock *rs, struct msghdr *msghdr)
  407. {
  408. struct rds_notifier *notifier;
  409. struct rds_rdma_notify cmsg;
  410. unsigned int count = 0, max_messages = ~0U;
  411. unsigned long flags;
  412. LIST_HEAD(copy);
  413. int err = 0;
  414. memset(&cmsg, 0, sizeof(cmsg)); /* fill holes with zero */
  415. /* put_cmsg copies to user space and thus may sleep. We can't do this
  416. * with rs_lock held, so first grab as many notifications as we can stuff
  417. * in the user provided cmsg buffer. We don't try to copy more, to avoid
  418. * losing notifications - except when the buffer is so small that it wouldn't
  419. * even hold a single notification. Then we give him as much of this single
  420. * msg as we can squeeze in, and set MSG_CTRUNC.
  421. */
  422. if (msghdr) {
  423. max_messages = msghdr->msg_controllen / CMSG_SPACE(sizeof(cmsg));
  424. if (!max_messages)
  425. max_messages = 1;
  426. }
  427. spin_lock_irqsave(&rs->rs_lock, flags);
  428. while (!list_empty(&rs->rs_notify_queue) && count < max_messages) {
  429. notifier = list_entry(rs->rs_notify_queue.next,
  430. struct rds_notifier, n_list);
  431. list_move(&notifier->n_list, &copy);
  432. count++;
  433. }
  434. spin_unlock_irqrestore(&rs->rs_lock, flags);
  435. if (!count)
  436. return 0;
  437. while (!list_empty(&copy)) {
  438. notifier = list_entry(copy.next, struct rds_notifier, n_list);
  439. if (msghdr) {
  440. cmsg.user_token = notifier->n_user_token;
  441. cmsg.status = notifier->n_status;
  442. err = put_cmsg(msghdr, SOL_RDS, RDS_CMSG_RDMA_STATUS,
  443. sizeof(cmsg), &cmsg);
  444. if (err)
  445. break;
  446. }
  447. list_del_init(&notifier->n_list);
  448. kfree(notifier);
  449. }
  450. /* If we bailed out because of an error in put_cmsg,
  451. * we may be left with one or more notifications that we
  452. * didn't process. Return them to the head of the list. */
  453. if (!list_empty(&copy)) {
  454. spin_lock_irqsave(&rs->rs_lock, flags);
  455. list_splice(&copy, &rs->rs_notify_queue);
  456. spin_unlock_irqrestore(&rs->rs_lock, flags);
  457. }
  458. return err;
  459. }
  460. /*
  461. * Queue a congestion notification
  462. */
  463. static int rds_notify_cong(struct rds_sock *rs, struct msghdr *msghdr)
  464. {
  465. uint64_t notify = rs->rs_cong_notify;
  466. unsigned long flags;
  467. int err;
  468. err = put_cmsg(msghdr, SOL_RDS, RDS_CMSG_CONG_UPDATE,
  469. sizeof(notify), &notify);
  470. if (err)
  471. return err;
  472. spin_lock_irqsave(&rs->rs_lock, flags);
  473. rs->rs_cong_notify &= ~notify;
  474. spin_unlock_irqrestore(&rs->rs_lock, flags);
  475. return 0;
  476. }
  477. /*
  478. * Receive any control messages.
  479. */
  480. static int rds_cmsg_recv(struct rds_incoming *inc, struct msghdr *msg,
  481. struct rds_sock *rs)
  482. {
  483. int ret = 0;
  484. if (inc->i_usercopy.rdma_cookie) {
  485. ret = put_cmsg(msg, SOL_RDS, RDS_CMSG_RDMA_DEST,
  486. sizeof(inc->i_usercopy.rdma_cookie),
  487. &inc->i_usercopy.rdma_cookie);
  488. if (ret)
  489. goto out;
  490. }
  491. if ((inc->i_usercopy.rx_tstamp != 0) &&
  492. sock_flag(rds_rs_to_sk(rs), SOCK_RCVTSTAMP)) {
  493. struct __kernel_old_timeval tv =
  494. ns_to_kernel_old_timeval(inc->i_usercopy.rx_tstamp);
  495. if (!sock_flag(rds_rs_to_sk(rs), SOCK_TSTAMP_NEW)) {
  496. ret = put_cmsg(msg, SOL_SOCKET, SO_TIMESTAMP_OLD,
  497. sizeof(tv), &tv);
  498. } else {
  499. struct __kernel_sock_timeval sk_tv;
  500. sk_tv.tv_sec = tv.tv_sec;
  501. sk_tv.tv_usec = tv.tv_usec;
  502. ret = put_cmsg(msg, SOL_SOCKET, SO_TIMESTAMP_NEW,
  503. sizeof(sk_tv), &sk_tv);
  504. }
  505. if (ret)
  506. goto out;
  507. }
  508. if (rs->rs_rx_traces) {
  509. struct rds_cmsg_rx_trace t;
  510. int i, j;
  511. memset(&t, 0, sizeof(t));
  512. inc->i_rx_lat_trace[RDS_MSG_RX_CMSG] = local_clock();
  513. t.rx_traces = rs->rs_rx_traces;
  514. for (i = 0; i < rs->rs_rx_traces; i++) {
  515. j = rs->rs_rx_trace[i];
  516. t.rx_trace_pos[i] = j;
  517. t.rx_trace[i] = inc->i_rx_lat_trace[j + 1] -
  518. inc->i_rx_lat_trace[j];
  519. }
  520. ret = put_cmsg(msg, SOL_RDS, RDS_CMSG_RXPATH_LATENCY,
  521. sizeof(t), &t);
  522. if (ret)
  523. goto out;
  524. }
  525. out:
  526. return ret;
  527. }
  528. static bool rds_recvmsg_zcookie(struct rds_sock *rs, struct msghdr *msg)
  529. {
  530. struct rds_msg_zcopy_queue *q = &rs->rs_zcookie_queue;
  531. struct rds_msg_zcopy_info *info = NULL;
  532. struct rds_zcopy_cookies *done;
  533. unsigned long flags;
  534. if (!msg->msg_control)
  535. return false;
  536. if (!sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY) ||
  537. msg->msg_controllen < CMSG_SPACE(sizeof(*done)))
  538. return false;
  539. spin_lock_irqsave(&q->lock, flags);
  540. if (!list_empty(&q->zcookie_head)) {
  541. info = list_entry(q->zcookie_head.next,
  542. struct rds_msg_zcopy_info, rs_zcookie_next);
  543. list_del(&info->rs_zcookie_next);
  544. }
  545. spin_unlock_irqrestore(&q->lock, flags);
  546. if (!info)
  547. return false;
  548. done = &info->zcookies;
  549. if (put_cmsg(msg, SOL_RDS, RDS_CMSG_ZCOPY_COMPLETION, sizeof(*done),
  550. done)) {
  551. spin_lock_irqsave(&q->lock, flags);
  552. list_add(&info->rs_zcookie_next, &q->zcookie_head);
  553. spin_unlock_irqrestore(&q->lock, flags);
  554. return false;
  555. }
  556. kfree(info);
  557. return true;
  558. }
  559. int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
  560. int msg_flags)
  561. {
  562. struct sock *sk = sock->sk;
  563. struct rds_sock *rs = rds_sk_to_rs(sk);
  564. long timeo;
  565. int ret = 0, nonblock = msg_flags & MSG_DONTWAIT;
  566. DECLARE_SOCKADDR(struct sockaddr_in6 *, sin6, msg->msg_name);
  567. DECLARE_SOCKADDR(struct sockaddr_in *, sin, msg->msg_name);
  568. struct rds_incoming *inc = NULL;
  569. /* udp_recvmsg()->sock_recvtimeo() gets away without locking too.. */
  570. timeo = sock_rcvtimeo(sk, nonblock);
  571. rdsdebug("size %zu flags 0x%x timeo %ld\n", size, msg_flags, timeo);
  572. if (msg_flags & MSG_OOB)
  573. goto out;
  574. if (msg_flags & MSG_ERRQUEUE)
  575. return sock_recv_errqueue(sk, msg, size, SOL_IP, IP_RECVERR);
  576. while (1) {
  577. /* If there are pending notifications, do those - and nothing else */
  578. if (!list_empty(&rs->rs_notify_queue)) {
  579. ret = rds_notify_queue_get(rs, msg);
  580. break;
  581. }
  582. if (rs->rs_cong_notify) {
  583. ret = rds_notify_cong(rs, msg);
  584. break;
  585. }
  586. if (!rds_next_incoming(rs, &inc)) {
  587. if (nonblock) {
  588. bool reaped = rds_recvmsg_zcookie(rs, msg);
  589. ret = reaped ? 0 : -EAGAIN;
  590. break;
  591. }
  592. timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
  593. (!list_empty(&rs->rs_notify_queue) ||
  594. rs->rs_cong_notify ||
  595. rds_next_incoming(rs, &inc)), timeo);
  596. rdsdebug("recvmsg woke inc %p timeo %ld\n", inc,
  597. timeo);
  598. if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
  599. continue;
  600. ret = timeo;
  601. if (ret == 0)
  602. ret = -ETIMEDOUT;
  603. break;
  604. }
  605. rdsdebug("copying inc %p from %pI6c:%u to user\n", inc,
  606. &inc->i_conn->c_faddr,
  607. ntohs(inc->i_hdr.h_sport));
  608. ret = inc->i_conn->c_trans->inc_copy_to_user(inc, &msg->msg_iter);
  609. if (ret < 0)
  610. break;
  611. /*
  612. * if the message we just copied isn't at the head of the
  613. * recv queue then someone else raced us to return it, try
  614. * to get the next message.
  615. */
  616. if (!rds_still_queued(rs, inc, !(msg_flags & MSG_PEEK))) {
  617. rds_inc_put(inc);
  618. inc = NULL;
  619. rds_stats_inc(s_recv_deliver_raced);
  620. iov_iter_revert(&msg->msg_iter, ret);
  621. continue;
  622. }
  623. if (ret < be32_to_cpu(inc->i_hdr.h_len)) {
  624. if (msg_flags & MSG_TRUNC)
  625. ret = be32_to_cpu(inc->i_hdr.h_len);
  626. msg->msg_flags |= MSG_TRUNC;
  627. }
  628. if (rds_cmsg_recv(inc, msg, rs)) {
  629. ret = -EFAULT;
  630. break;
  631. }
  632. rds_recvmsg_zcookie(rs, msg);
  633. rds_stats_inc(s_recv_delivered);
  634. if (msg->msg_name) {
  635. if (ipv6_addr_v4mapped(&inc->i_saddr)) {
  636. sin->sin_family = AF_INET;
  637. sin->sin_port = inc->i_hdr.h_sport;
  638. sin->sin_addr.s_addr =
  639. inc->i_saddr.s6_addr32[3];
  640. memset(sin->sin_zero, 0, sizeof(sin->sin_zero));
  641. msg->msg_namelen = sizeof(*sin);
  642. } else {
  643. sin6->sin6_family = AF_INET6;
  644. sin6->sin6_port = inc->i_hdr.h_sport;
  645. sin6->sin6_addr = inc->i_saddr;
  646. sin6->sin6_flowinfo = 0;
  647. sin6->sin6_scope_id = rs->rs_bound_scope_id;
  648. msg->msg_namelen = sizeof(*sin6);
  649. }
  650. }
  651. break;
  652. }
  653. if (inc)
  654. rds_inc_put(inc);
  655. out:
  656. return ret;
  657. }
  658. /*
  659. * The socket is being shut down and we're asked to drop messages that were
  660. * queued for recvmsg. The caller has unbound the socket so the receive path
  661. * won't queue any more incoming fragments or messages on the socket.
  662. */
  663. void rds_clear_recv_queue(struct rds_sock *rs)
  664. {
  665. struct sock *sk = rds_rs_to_sk(rs);
  666. struct rds_incoming *inc, *tmp;
  667. unsigned long flags;
  668. write_lock_irqsave(&rs->rs_recv_lock, flags);
  669. list_for_each_entry_safe(inc, tmp, &rs->rs_recv_queue, i_item) {
  670. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  671. -be32_to_cpu(inc->i_hdr.h_len),
  672. inc->i_hdr.h_dport);
  673. list_del_init(&inc->i_item);
  674. rds_inc_put(inc);
  675. }
  676. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  677. }
  678. /*
  679. * inc->i_saddr isn't used here because it is only set in the receive
  680. * path.
  681. */
  682. void rds_inc_info_copy(struct rds_incoming *inc,
  683. struct rds_info_iterator *iter,
  684. __be32 saddr, __be32 daddr, int flip)
  685. {
  686. struct rds_info_message minfo;
  687. minfo.seq = be64_to_cpu(inc->i_hdr.h_sequence);
  688. minfo.len = be32_to_cpu(inc->i_hdr.h_len);
  689. minfo.tos = inc->i_conn->c_tos;
  690. if (flip) {
  691. minfo.laddr = daddr;
  692. minfo.faddr = saddr;
  693. minfo.lport = inc->i_hdr.h_dport;
  694. minfo.fport = inc->i_hdr.h_sport;
  695. } else {
  696. minfo.laddr = saddr;
  697. minfo.faddr = daddr;
  698. minfo.lport = inc->i_hdr.h_sport;
  699. minfo.fport = inc->i_hdr.h_dport;
  700. }
  701. minfo.flags = 0;
  702. rds_info_copy(iter, &minfo, sizeof(minfo));
  703. }
  704. #if IS_ENABLED(CONFIG_IPV6)
  705. void rds6_inc_info_copy(struct rds_incoming *inc,
  706. struct rds_info_iterator *iter,
  707. struct in6_addr *saddr, struct in6_addr *daddr,
  708. int flip)
  709. {
  710. struct rds6_info_message minfo6;
  711. minfo6.seq = be64_to_cpu(inc->i_hdr.h_sequence);
  712. minfo6.len = be32_to_cpu(inc->i_hdr.h_len);
  713. minfo6.tos = inc->i_conn->c_tos;
  714. if (flip) {
  715. minfo6.laddr = *daddr;
  716. minfo6.faddr = *saddr;
  717. minfo6.lport = inc->i_hdr.h_dport;
  718. minfo6.fport = inc->i_hdr.h_sport;
  719. } else {
  720. minfo6.laddr = *saddr;
  721. minfo6.faddr = *daddr;
  722. minfo6.lport = inc->i_hdr.h_sport;
  723. minfo6.fport = inc->i_hdr.h_dport;
  724. }
  725. minfo6.flags = 0;
  726. rds_info_copy(iter, &minfo6, sizeof(minfo6));
  727. }
  728. #endif