recv.c 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701
  1. /*
  2. * Copyright (c) 2006 Oracle. 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. __be32 saddr)
  43. {
  44. atomic_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_rdma_cookie = 0;
  49. inc->i_rx_tstamp.tv_sec = 0;
  50. inc->i_rx_tstamp.tv_usec = 0;
  51. }
  52. EXPORT_SYMBOL_GPL(rds_inc_init);
  53. void rds_inc_path_init(struct rds_incoming *inc, struct rds_conn_path *cp,
  54. __be32 saddr)
  55. {
  56. atomic_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_rdma_cookie = 0;
  62. inc->i_rx_tstamp.tv_sec = 0;
  63. inc->i_rx_tstamp.tv_usec = 0;
  64. }
  65. EXPORT_SYMBOL_GPL(rds_inc_path_init);
  66. static void rds_inc_addref(struct rds_incoming *inc)
  67. {
  68. rdsdebug("addref inc %p ref %d\n", inc, atomic_read(&inc->i_refcount));
  69. atomic_inc(&inc->i_refcount);
  70. }
  71. void rds_inc_put(struct rds_incoming *inc)
  72. {
  73. rdsdebug("put inc %p ref %d\n", inc, atomic_read(&inc->i_refcount));
  74. if (atomic_dec_and_test(&inc->i_refcount)) {
  75. BUG_ON(!list_empty(&inc->i_item));
  76. inc->i_conn->c_trans->inc_free(inc);
  77. }
  78. }
  79. EXPORT_SYMBOL_GPL(rds_inc_put);
  80. static void rds_recv_rcvbuf_delta(struct rds_sock *rs, struct sock *sk,
  81. struct rds_cong_map *map,
  82. int delta, __be16 port)
  83. {
  84. int now_congested;
  85. if (delta == 0)
  86. return;
  87. rs->rs_rcv_bytes += delta;
  88. now_congested = rs->rs_rcv_bytes > rds_sk_rcvbuf(rs);
  89. rdsdebug("rs %p (%pI4:%u) recv bytes %d buf %d "
  90. "now_cong %d delta %d\n",
  91. rs, &rs->rs_bound_addr,
  92. ntohs(rs->rs_bound_port), rs->rs_rcv_bytes,
  93. rds_sk_rcvbuf(rs), now_congested, delta);
  94. /* wasn't -> am congested */
  95. if (!rs->rs_congested && now_congested) {
  96. rs->rs_congested = 1;
  97. rds_cong_set_bit(map, port);
  98. rds_cong_queue_updates(map);
  99. }
  100. /* was -> aren't congested */
  101. /* Require more free space before reporting uncongested to prevent
  102. bouncing cong/uncong state too often */
  103. else if (rs->rs_congested && (rs->rs_rcv_bytes < (rds_sk_rcvbuf(rs)/2))) {
  104. rs->rs_congested = 0;
  105. rds_cong_clear_bit(map, port);
  106. rds_cong_queue_updates(map);
  107. }
  108. /* do nothing if no change in cong state */
  109. }
  110. static void rds_conn_peer_gen_update(struct rds_connection *conn,
  111. u32 peer_gen_num)
  112. {
  113. int i;
  114. struct rds_message *rm, *tmp;
  115. unsigned long flags;
  116. WARN_ON(conn->c_trans->t_type != RDS_TRANS_TCP);
  117. if (peer_gen_num != 0) {
  118. if (conn->c_peer_gen_num != 0 &&
  119. peer_gen_num != conn->c_peer_gen_num) {
  120. for (i = 0; i < RDS_MPATH_WORKERS; i++) {
  121. struct rds_conn_path *cp;
  122. cp = &conn->c_path[i];
  123. spin_lock_irqsave(&cp->cp_lock, flags);
  124. cp->cp_next_tx_seq = 1;
  125. cp->cp_next_rx_seq = 0;
  126. list_for_each_entry_safe(rm, tmp,
  127. &cp->cp_retrans,
  128. m_conn_item) {
  129. set_bit(RDS_MSG_FLUSH, &rm->m_flags);
  130. }
  131. spin_unlock_irqrestore(&cp->cp_lock, flags);
  132. }
  133. }
  134. conn->c_peer_gen_num = peer_gen_num;
  135. }
  136. }
  137. /*
  138. * Process all extension headers that come with this message.
  139. */
  140. static void rds_recv_incoming_exthdrs(struct rds_incoming *inc, struct rds_sock *rs)
  141. {
  142. struct rds_header *hdr = &inc->i_hdr;
  143. unsigned int pos = 0, type, len;
  144. union {
  145. struct rds_ext_header_version version;
  146. struct rds_ext_header_rdma rdma;
  147. struct rds_ext_header_rdma_dest rdma_dest;
  148. } buffer;
  149. while (1) {
  150. len = sizeof(buffer);
  151. type = rds_message_next_extension(hdr, &pos, &buffer, &len);
  152. if (type == RDS_EXTHDR_NONE)
  153. break;
  154. /* Process extension header here */
  155. switch (type) {
  156. case RDS_EXTHDR_RDMA:
  157. rds_rdma_unuse(rs, be32_to_cpu(buffer.rdma.h_rdma_rkey), 0);
  158. break;
  159. case RDS_EXTHDR_RDMA_DEST:
  160. /* We ignore the size for now. We could stash it
  161. * somewhere and use it for error checking. */
  162. inc->i_rdma_cookie = rds_rdma_make_cookie(
  163. be32_to_cpu(buffer.rdma_dest.h_rdma_rkey),
  164. be32_to_cpu(buffer.rdma_dest.h_rdma_offset));
  165. break;
  166. }
  167. }
  168. }
  169. static void rds_recv_hs_exthdrs(struct rds_header *hdr,
  170. struct rds_connection *conn)
  171. {
  172. unsigned int pos = 0, type, len;
  173. union {
  174. struct rds_ext_header_version version;
  175. u16 rds_npaths;
  176. u32 rds_gen_num;
  177. } buffer;
  178. u32 new_peer_gen_num = 0;
  179. while (1) {
  180. len = sizeof(buffer);
  181. type = rds_message_next_extension(hdr, &pos, &buffer, &len);
  182. if (type == RDS_EXTHDR_NONE)
  183. break;
  184. /* Process extension header here */
  185. switch (type) {
  186. case RDS_EXTHDR_NPATHS:
  187. conn->c_npaths = min_t(int, RDS_MPATH_WORKERS,
  188. buffer.rds_npaths);
  189. break;
  190. case RDS_EXTHDR_GEN_NUM:
  191. new_peer_gen_num = buffer.rds_gen_num;
  192. break;
  193. default:
  194. pr_warn_ratelimited("ignoring unknown exthdr type "
  195. "0x%x\n", type);
  196. }
  197. }
  198. /* if RDS_EXTHDR_NPATHS was not found, default to a single-path */
  199. conn->c_npaths = max_t(int, conn->c_npaths, 1);
  200. rds_conn_peer_gen_update(conn, new_peer_gen_num);
  201. }
  202. /* rds_start_mprds() will synchronously start multiple paths when appropriate.
  203. * The scheme is based on the following rules:
  204. *
  205. * 1. rds_sendmsg on first connect attempt sends the probe ping, with the
  206. * sender's npaths (s_npaths)
  207. * 2. rcvr of probe-ping knows the mprds_paths = min(s_npaths, r_npaths). It
  208. * sends back a probe-pong with r_npaths. After that, if rcvr is the
  209. * smaller ip addr, it starts rds_conn_path_connect_if_down on all
  210. * mprds_paths.
  211. * 3. sender gets woken up, and can move to rds_conn_path_connect_if_down.
  212. * If it is the smaller ipaddr, rds_conn_path_connect_if_down can be
  213. * called after reception of the probe-pong on all mprds_paths.
  214. * Otherwise (sender of probe-ping is not the smaller ip addr): just call
  215. * rds_conn_path_connect_if_down on the hashed path. (see rule 4)
  216. * 4. when cp_index > 0, rds_connect_worker must only trigger
  217. * a connection if laddr < faddr.
  218. * 5. sender may end up queuing the packet on the cp. will get sent out later.
  219. * when connection is completed.
  220. */
  221. static void rds_start_mprds(struct rds_connection *conn)
  222. {
  223. int i;
  224. struct rds_conn_path *cp;
  225. if (conn->c_npaths > 1 && conn->c_laddr < conn->c_faddr) {
  226. for (i = 1; i < conn->c_npaths; i++) {
  227. cp = &conn->c_path[i];
  228. rds_conn_path_connect_if_down(cp);
  229. }
  230. }
  231. }
  232. /*
  233. * The transport must make sure that this is serialized against other
  234. * rx and conn reset on this specific conn.
  235. *
  236. * We currently assert that only one fragmented message will be sent
  237. * down a connection at a time. This lets us reassemble in the conn
  238. * instead of per-flow which means that we don't have to go digging through
  239. * flows to tear down partial reassembly progress on conn failure and
  240. * we save flow lookup and locking for each frag arrival. It does mean
  241. * that small messages will wait behind large ones. Fragmenting at all
  242. * is only to reduce the memory consumption of pre-posted buffers.
  243. *
  244. * The caller passes in saddr and daddr instead of us getting it from the
  245. * conn. This lets loopback, who only has one conn for both directions,
  246. * tell us which roles the addrs in the conn are playing for this message.
  247. */
  248. void rds_recv_incoming(struct rds_connection *conn, __be32 saddr, __be32 daddr,
  249. struct rds_incoming *inc, gfp_t gfp)
  250. {
  251. struct rds_sock *rs = NULL;
  252. struct sock *sk;
  253. unsigned long flags;
  254. struct rds_conn_path *cp;
  255. inc->i_conn = conn;
  256. inc->i_rx_jiffies = jiffies;
  257. if (conn->c_trans->t_mp_capable)
  258. cp = inc->i_conn_path;
  259. else
  260. cp = &conn->c_path[0];
  261. rdsdebug("conn %p next %llu inc %p seq %llu len %u sport %u dport %u "
  262. "flags 0x%x rx_jiffies %lu\n", conn,
  263. (unsigned long long)cp->cp_next_rx_seq,
  264. inc,
  265. (unsigned long long)be64_to_cpu(inc->i_hdr.h_sequence),
  266. be32_to_cpu(inc->i_hdr.h_len),
  267. be16_to_cpu(inc->i_hdr.h_sport),
  268. be16_to_cpu(inc->i_hdr.h_dport),
  269. inc->i_hdr.h_flags,
  270. inc->i_rx_jiffies);
  271. /*
  272. * Sequence numbers should only increase. Messages get their
  273. * sequence number as they're queued in a sending conn. They
  274. * can be dropped, though, if the sending socket is closed before
  275. * they hit the wire. So sequence numbers can skip forward
  276. * under normal operation. They can also drop back in the conn
  277. * failover case as previously sent messages are resent down the
  278. * new instance of a conn. We drop those, otherwise we have
  279. * to assume that the next valid seq does not come after a
  280. * hole in the fragment stream.
  281. *
  282. * The headers don't give us a way to realize if fragments of
  283. * a message have been dropped. We assume that frags that arrive
  284. * to a flow are part of the current message on the flow that is
  285. * being reassembled. This means that senders can't drop messages
  286. * from the sending conn until all their frags are sent.
  287. *
  288. * XXX we could spend more on the wire to get more robust failure
  289. * detection, arguably worth it to avoid data corruption.
  290. */
  291. if (be64_to_cpu(inc->i_hdr.h_sequence) < cp->cp_next_rx_seq &&
  292. (inc->i_hdr.h_flags & RDS_FLAG_RETRANSMITTED)) {
  293. rds_stats_inc(s_recv_drop_old_seq);
  294. goto out;
  295. }
  296. cp->cp_next_rx_seq = be64_to_cpu(inc->i_hdr.h_sequence) + 1;
  297. if (rds_sysctl_ping_enable && inc->i_hdr.h_dport == 0) {
  298. if (inc->i_hdr.h_sport == 0) {
  299. rdsdebug("ignore ping with 0 sport from 0x%x\n", saddr);
  300. goto out;
  301. }
  302. rds_stats_inc(s_recv_ping);
  303. rds_send_pong(cp, inc->i_hdr.h_sport);
  304. /* if this is a handshake ping, start multipath if necessary */
  305. if (RDS_HS_PROBE(inc->i_hdr.h_sport, inc->i_hdr.h_dport)) {
  306. rds_recv_hs_exthdrs(&inc->i_hdr, cp->cp_conn);
  307. rds_start_mprds(cp->cp_conn);
  308. }
  309. goto out;
  310. }
  311. if (inc->i_hdr.h_dport == RDS_FLAG_PROBE_PORT &&
  312. inc->i_hdr.h_sport == 0) {
  313. rds_recv_hs_exthdrs(&inc->i_hdr, cp->cp_conn);
  314. /* if this is a handshake pong, start multipath if necessary */
  315. rds_start_mprds(cp->cp_conn);
  316. wake_up(&cp->cp_conn->c_hs_waitq);
  317. goto out;
  318. }
  319. rs = rds_find_bound(daddr, inc->i_hdr.h_dport);
  320. if (!rs) {
  321. rds_stats_inc(s_recv_drop_no_sock);
  322. goto out;
  323. }
  324. /* Process extension headers */
  325. rds_recv_incoming_exthdrs(inc, rs);
  326. /* We can be racing with rds_release() which marks the socket dead. */
  327. sk = rds_rs_to_sk(rs);
  328. /* serialize with rds_release -> sock_orphan */
  329. write_lock_irqsave(&rs->rs_recv_lock, flags);
  330. if (!sock_flag(sk, SOCK_DEAD)) {
  331. rdsdebug("adding inc %p to rs %p's recv queue\n", inc, rs);
  332. rds_stats_inc(s_recv_queued);
  333. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  334. be32_to_cpu(inc->i_hdr.h_len),
  335. inc->i_hdr.h_dport);
  336. if (sock_flag(sk, SOCK_RCVTSTAMP))
  337. do_gettimeofday(&inc->i_rx_tstamp);
  338. rds_inc_addref(inc);
  339. list_add_tail(&inc->i_item, &rs->rs_recv_queue);
  340. __rds_wake_sk_sleep(sk);
  341. } else {
  342. rds_stats_inc(s_recv_drop_dead_sock);
  343. }
  344. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  345. out:
  346. if (rs)
  347. rds_sock_put(rs);
  348. }
  349. EXPORT_SYMBOL_GPL(rds_recv_incoming);
  350. /*
  351. * be very careful here. This is being called as the condition in
  352. * wait_event_*() needs to cope with being called many times.
  353. */
  354. static int rds_next_incoming(struct rds_sock *rs, struct rds_incoming **inc)
  355. {
  356. unsigned long flags;
  357. if (!*inc) {
  358. read_lock_irqsave(&rs->rs_recv_lock, flags);
  359. if (!list_empty(&rs->rs_recv_queue)) {
  360. *inc = list_entry(rs->rs_recv_queue.next,
  361. struct rds_incoming,
  362. i_item);
  363. rds_inc_addref(*inc);
  364. }
  365. read_unlock_irqrestore(&rs->rs_recv_lock, flags);
  366. }
  367. return *inc != NULL;
  368. }
  369. static int rds_still_queued(struct rds_sock *rs, struct rds_incoming *inc,
  370. int drop)
  371. {
  372. struct sock *sk = rds_rs_to_sk(rs);
  373. int ret = 0;
  374. unsigned long flags;
  375. write_lock_irqsave(&rs->rs_recv_lock, flags);
  376. if (!list_empty(&inc->i_item)) {
  377. ret = 1;
  378. if (drop) {
  379. /* XXX make sure this i_conn is reliable */
  380. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  381. -be32_to_cpu(inc->i_hdr.h_len),
  382. inc->i_hdr.h_dport);
  383. list_del_init(&inc->i_item);
  384. rds_inc_put(inc);
  385. }
  386. }
  387. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  388. rdsdebug("inc %p rs %p still %d dropped %d\n", inc, rs, ret, drop);
  389. return ret;
  390. }
  391. /*
  392. * Pull errors off the error queue.
  393. * If msghdr is NULL, we will just purge the error queue.
  394. */
  395. int rds_notify_queue_get(struct rds_sock *rs, struct msghdr *msghdr)
  396. {
  397. struct rds_notifier *notifier;
  398. struct rds_rdma_notify cmsg = { 0 }; /* fill holes with zero */
  399. unsigned int count = 0, max_messages = ~0U;
  400. unsigned long flags;
  401. LIST_HEAD(copy);
  402. int err = 0;
  403. /* put_cmsg copies to user space and thus may sleep. We can't do this
  404. * with rs_lock held, so first grab as many notifications as we can stuff
  405. * in the user provided cmsg buffer. We don't try to copy more, to avoid
  406. * losing notifications - except when the buffer is so small that it wouldn't
  407. * even hold a single notification. Then we give him as much of this single
  408. * msg as we can squeeze in, and set MSG_CTRUNC.
  409. */
  410. if (msghdr) {
  411. max_messages = msghdr->msg_controllen / CMSG_SPACE(sizeof(cmsg));
  412. if (!max_messages)
  413. max_messages = 1;
  414. }
  415. spin_lock_irqsave(&rs->rs_lock, flags);
  416. while (!list_empty(&rs->rs_notify_queue) && count < max_messages) {
  417. notifier = list_entry(rs->rs_notify_queue.next,
  418. struct rds_notifier, n_list);
  419. list_move(&notifier->n_list, &copy);
  420. count++;
  421. }
  422. spin_unlock_irqrestore(&rs->rs_lock, flags);
  423. if (!count)
  424. return 0;
  425. while (!list_empty(&copy)) {
  426. notifier = list_entry(copy.next, struct rds_notifier, n_list);
  427. if (msghdr) {
  428. cmsg.user_token = notifier->n_user_token;
  429. cmsg.status = notifier->n_status;
  430. err = put_cmsg(msghdr, SOL_RDS, RDS_CMSG_RDMA_STATUS,
  431. sizeof(cmsg), &cmsg);
  432. if (err)
  433. break;
  434. }
  435. list_del_init(&notifier->n_list);
  436. kfree(notifier);
  437. }
  438. /* If we bailed out because of an error in put_cmsg,
  439. * we may be left with one or more notifications that we
  440. * didn't process. Return them to the head of the list. */
  441. if (!list_empty(&copy)) {
  442. spin_lock_irqsave(&rs->rs_lock, flags);
  443. list_splice(&copy, &rs->rs_notify_queue);
  444. spin_unlock_irqrestore(&rs->rs_lock, flags);
  445. }
  446. return err;
  447. }
  448. /*
  449. * Queue a congestion notification
  450. */
  451. static int rds_notify_cong(struct rds_sock *rs, struct msghdr *msghdr)
  452. {
  453. uint64_t notify = rs->rs_cong_notify;
  454. unsigned long flags;
  455. int err;
  456. err = put_cmsg(msghdr, SOL_RDS, RDS_CMSG_CONG_UPDATE,
  457. sizeof(notify), &notify);
  458. if (err)
  459. return err;
  460. spin_lock_irqsave(&rs->rs_lock, flags);
  461. rs->rs_cong_notify &= ~notify;
  462. spin_unlock_irqrestore(&rs->rs_lock, flags);
  463. return 0;
  464. }
  465. /*
  466. * Receive any control messages.
  467. */
  468. static int rds_cmsg_recv(struct rds_incoming *inc, struct msghdr *msg,
  469. struct rds_sock *rs)
  470. {
  471. int ret = 0;
  472. if (inc->i_rdma_cookie) {
  473. ret = put_cmsg(msg, SOL_RDS, RDS_CMSG_RDMA_DEST,
  474. sizeof(inc->i_rdma_cookie), &inc->i_rdma_cookie);
  475. if (ret)
  476. return ret;
  477. }
  478. if ((inc->i_rx_tstamp.tv_sec != 0) &&
  479. sock_flag(rds_rs_to_sk(rs), SOCK_RCVTSTAMP)) {
  480. ret = put_cmsg(msg, SOL_SOCKET, SCM_TIMESTAMP,
  481. sizeof(struct timeval),
  482. &inc->i_rx_tstamp);
  483. if (ret)
  484. return ret;
  485. }
  486. return 0;
  487. }
  488. int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
  489. int msg_flags)
  490. {
  491. struct sock *sk = sock->sk;
  492. struct rds_sock *rs = rds_sk_to_rs(sk);
  493. long timeo;
  494. int ret = 0, nonblock = msg_flags & MSG_DONTWAIT;
  495. DECLARE_SOCKADDR(struct sockaddr_in *, sin, msg->msg_name);
  496. struct rds_incoming *inc = NULL;
  497. /* udp_recvmsg()->sock_recvtimeo() gets away without locking too.. */
  498. timeo = sock_rcvtimeo(sk, nonblock);
  499. rdsdebug("size %zu flags 0x%x timeo %ld\n", size, msg_flags, timeo);
  500. if (msg_flags & MSG_OOB)
  501. goto out;
  502. while (1) {
  503. struct iov_iter save;
  504. /* If there are pending notifications, do those - and nothing else */
  505. if (!list_empty(&rs->rs_notify_queue)) {
  506. ret = rds_notify_queue_get(rs, msg);
  507. break;
  508. }
  509. if (rs->rs_cong_notify) {
  510. ret = rds_notify_cong(rs, msg);
  511. break;
  512. }
  513. if (!rds_next_incoming(rs, &inc)) {
  514. if (nonblock) {
  515. ret = -EAGAIN;
  516. break;
  517. }
  518. timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
  519. (!list_empty(&rs->rs_notify_queue) ||
  520. rs->rs_cong_notify ||
  521. rds_next_incoming(rs, &inc)), timeo);
  522. rdsdebug("recvmsg woke inc %p timeo %ld\n", inc,
  523. timeo);
  524. if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
  525. continue;
  526. ret = timeo;
  527. if (ret == 0)
  528. ret = -ETIMEDOUT;
  529. break;
  530. }
  531. rdsdebug("copying inc %p from %pI4:%u to user\n", inc,
  532. &inc->i_conn->c_faddr,
  533. ntohs(inc->i_hdr.h_sport));
  534. save = msg->msg_iter;
  535. ret = inc->i_conn->c_trans->inc_copy_to_user(inc, &msg->msg_iter);
  536. if (ret < 0)
  537. break;
  538. /*
  539. * if the message we just copied isn't at the head of the
  540. * recv queue then someone else raced us to return it, try
  541. * to get the next message.
  542. */
  543. if (!rds_still_queued(rs, inc, !(msg_flags & MSG_PEEK))) {
  544. rds_inc_put(inc);
  545. inc = NULL;
  546. rds_stats_inc(s_recv_deliver_raced);
  547. msg->msg_iter = save;
  548. continue;
  549. }
  550. if (ret < be32_to_cpu(inc->i_hdr.h_len)) {
  551. if (msg_flags & MSG_TRUNC)
  552. ret = be32_to_cpu(inc->i_hdr.h_len);
  553. msg->msg_flags |= MSG_TRUNC;
  554. }
  555. if (rds_cmsg_recv(inc, msg, rs)) {
  556. ret = -EFAULT;
  557. goto out;
  558. }
  559. rds_stats_inc(s_recv_delivered);
  560. if (sin) {
  561. sin->sin_family = AF_INET;
  562. sin->sin_port = inc->i_hdr.h_sport;
  563. sin->sin_addr.s_addr = inc->i_saddr;
  564. memset(sin->sin_zero, 0, sizeof(sin->sin_zero));
  565. msg->msg_namelen = sizeof(*sin);
  566. }
  567. break;
  568. }
  569. if (inc)
  570. rds_inc_put(inc);
  571. out:
  572. return ret;
  573. }
  574. /*
  575. * The socket is being shut down and we're asked to drop messages that were
  576. * queued for recvmsg. The caller has unbound the socket so the receive path
  577. * won't queue any more incoming fragments or messages on the socket.
  578. */
  579. void rds_clear_recv_queue(struct rds_sock *rs)
  580. {
  581. struct sock *sk = rds_rs_to_sk(rs);
  582. struct rds_incoming *inc, *tmp;
  583. unsigned long flags;
  584. write_lock_irqsave(&rs->rs_recv_lock, flags);
  585. list_for_each_entry_safe(inc, tmp, &rs->rs_recv_queue, i_item) {
  586. rds_recv_rcvbuf_delta(rs, sk, inc->i_conn->c_lcong,
  587. -be32_to_cpu(inc->i_hdr.h_len),
  588. inc->i_hdr.h_dport);
  589. list_del_init(&inc->i_item);
  590. rds_inc_put(inc);
  591. }
  592. write_unlock_irqrestore(&rs->rs_recv_lock, flags);
  593. }
  594. /*
  595. * inc->i_saddr isn't used here because it is only set in the receive
  596. * path.
  597. */
  598. void rds_inc_info_copy(struct rds_incoming *inc,
  599. struct rds_info_iterator *iter,
  600. __be32 saddr, __be32 daddr, int flip)
  601. {
  602. struct rds_info_message minfo;
  603. minfo.seq = be64_to_cpu(inc->i_hdr.h_sequence);
  604. minfo.len = be32_to_cpu(inc->i_hdr.h_len);
  605. if (flip) {
  606. minfo.laddr = daddr;
  607. minfo.faddr = saddr;
  608. minfo.lport = inc->i_hdr.h_dport;
  609. minfo.fport = inc->i_hdr.h_sport;
  610. } else {
  611. minfo.laddr = saddr;
  612. minfo.faddr = daddr;
  613. minfo.lport = inc->i_hdr.h_sport;
  614. minfo.fport = inc->i_hdr.h_dport;
  615. }
  616. minfo.flags = 0;
  617. rds_info_copy(iter, &minfo, sizeof(minfo));
  618. }