recv.c 21 KB

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