send.c 38 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462
  1. /*
  2. * Copyright (c) 2006, 2018 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/moduleparam.h>
  35. #include <linux/gfp.h>
  36. #include <net/sock.h>
  37. #include <linux/in.h>
  38. #include <linux/list.h>
  39. #include <linux/ratelimit.h>
  40. #include <linux/export.h>
  41. #include <linux/sizes.h>
  42. #include "rds.h"
  43. /* When transmitting messages in rds_send_xmit, we need to emerge from
  44. * time to time and briefly release the CPU. Otherwise the softlock watchdog
  45. * will kick our shin.
  46. * Also, it seems fairer to not let one busy connection stall all the
  47. * others.
  48. *
  49. * send_batch_count is the number of times we'll loop in send_xmit. Setting
  50. * it to 0 will restore the old behavior (where we looped until we had
  51. * drained the queue).
  52. */
  53. static int send_batch_count = SZ_1K;
  54. module_param(send_batch_count, int, 0444);
  55. MODULE_PARM_DESC(send_batch_count, " batch factor when working the send queue");
  56. static void rds_send_remove_from_sock(struct list_head *messages, int status);
  57. /*
  58. * Reset the send state. Callers must ensure that this doesn't race with
  59. * rds_send_xmit().
  60. */
  61. void rds_send_path_reset(struct rds_conn_path *cp)
  62. {
  63. struct rds_message *rm, *tmp;
  64. unsigned long flags;
  65. if (cp->cp_xmit_rm) {
  66. rm = cp->cp_xmit_rm;
  67. cp->cp_xmit_rm = NULL;
  68. /* Tell the user the RDMA op is no longer mapped by the
  69. * transport. This isn't entirely true (it's flushed out
  70. * independently) but as the connection is down, there's
  71. * no ongoing RDMA to/from that memory */
  72. rds_message_unmapped(rm);
  73. rds_message_put(rm);
  74. }
  75. cp->cp_xmit_sg = 0;
  76. cp->cp_xmit_hdr_off = 0;
  77. cp->cp_xmit_data_off = 0;
  78. cp->cp_xmit_atomic_sent = 0;
  79. cp->cp_xmit_rdma_sent = 0;
  80. cp->cp_xmit_data_sent = 0;
  81. cp->cp_conn->c_map_queued = 0;
  82. cp->cp_unacked_packets = rds_sysctl_max_unacked_packets;
  83. cp->cp_unacked_bytes = rds_sysctl_max_unacked_bytes;
  84. /* Mark messages as retransmissions, and move them to the send q */
  85. spin_lock_irqsave(&cp->cp_lock, flags);
  86. list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
  87. set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
  88. set_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags);
  89. }
  90. list_splice_init(&cp->cp_retrans, &cp->cp_send_queue);
  91. spin_unlock_irqrestore(&cp->cp_lock, flags);
  92. }
  93. EXPORT_SYMBOL_GPL(rds_send_path_reset);
  94. static int acquire_in_xmit(struct rds_conn_path *cp)
  95. {
  96. return test_and_set_bit(RDS_IN_XMIT, &cp->cp_flags) == 0;
  97. }
  98. static void release_in_xmit(struct rds_conn_path *cp)
  99. {
  100. clear_bit(RDS_IN_XMIT, &cp->cp_flags);
  101. smp_mb__after_atomic();
  102. /*
  103. * We don't use wait_on_bit()/wake_up_bit() because our waking is in a
  104. * hot path and finding waiters is very rare. We don't want to walk
  105. * the system-wide hashed waitqueue buckets in the fast path only to
  106. * almost never find waiters.
  107. */
  108. if (waitqueue_active(&cp->cp_waitq))
  109. wake_up_all(&cp->cp_waitq);
  110. }
  111. /*
  112. * We're making the conscious trade-off here to only send one message
  113. * down the connection at a time.
  114. * Pro:
  115. * - tx queueing is a simple fifo list
  116. * - reassembly is optional and easily done by transports per conn
  117. * - no per flow rx lookup at all, straight to the socket
  118. * - less per-frag memory and wire overhead
  119. * Con:
  120. * - queued acks can be delayed behind large messages
  121. * Depends:
  122. * - small message latency is higher behind queued large messages
  123. * - large message latency isn't starved by intervening small sends
  124. */
  125. int rds_send_xmit(struct rds_conn_path *cp)
  126. {
  127. struct rds_connection *conn = cp->cp_conn;
  128. struct rds_message *rm;
  129. unsigned long flags;
  130. unsigned int tmp;
  131. struct scatterlist *sg;
  132. int ret = 0;
  133. LIST_HEAD(to_be_dropped);
  134. int batch_count;
  135. unsigned long send_gen = 0;
  136. restart:
  137. batch_count = 0;
  138. /*
  139. * sendmsg calls here after having queued its message on the send
  140. * queue. We only have one task feeding the connection at a time. If
  141. * another thread is already feeding the queue then we back off. This
  142. * avoids blocking the caller and trading per-connection data between
  143. * caches per message.
  144. */
  145. if (!acquire_in_xmit(cp)) {
  146. rds_stats_inc(s_send_lock_contention);
  147. ret = -ENOMEM;
  148. goto out;
  149. }
  150. if (rds_destroy_pending(cp->cp_conn)) {
  151. release_in_xmit(cp);
  152. ret = -ENETUNREACH; /* dont requeue send work */
  153. goto out;
  154. }
  155. /*
  156. * we record the send generation after doing the xmit acquire.
  157. * if someone else manages to jump in and do some work, we'll use
  158. * this to avoid a goto restart farther down.
  159. *
  160. * The acquire_in_xmit() check above ensures that only one
  161. * caller can increment c_send_gen at any time.
  162. */
  163. send_gen = READ_ONCE(cp->cp_send_gen) + 1;
  164. WRITE_ONCE(cp->cp_send_gen, send_gen);
  165. /*
  166. * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT,
  167. * we do the opposite to avoid races.
  168. */
  169. if (!rds_conn_path_up(cp)) {
  170. release_in_xmit(cp);
  171. ret = 0;
  172. goto out;
  173. }
  174. if (conn->c_trans->xmit_path_prepare)
  175. conn->c_trans->xmit_path_prepare(cp);
  176. /*
  177. * spin trying to push headers and data down the connection until
  178. * the connection doesn't make forward progress.
  179. */
  180. while (1) {
  181. rm = cp->cp_xmit_rm;
  182. /*
  183. * If between sending messages, we can send a pending congestion
  184. * map update.
  185. */
  186. if (!rm && test_and_clear_bit(0, &conn->c_map_queued)) {
  187. rm = rds_cong_update_alloc(conn);
  188. if (IS_ERR(rm)) {
  189. ret = PTR_ERR(rm);
  190. break;
  191. }
  192. rm->data.op_active = 1;
  193. rm->m_inc.i_conn_path = cp;
  194. rm->m_inc.i_conn = cp->cp_conn;
  195. cp->cp_xmit_rm = rm;
  196. }
  197. /*
  198. * If not already working on one, grab the next message.
  199. *
  200. * cp_xmit_rm holds a ref while we're sending this message down
  201. * the connction. We can use this ref while holding the
  202. * send_sem.. rds_send_reset() is serialized with it.
  203. */
  204. if (!rm) {
  205. unsigned int len;
  206. batch_count++;
  207. /* we want to process as big a batch as we can, but
  208. * we also want to avoid softlockups. If we've been
  209. * through a lot of messages, lets back off and see
  210. * if anyone else jumps in
  211. */
  212. if (batch_count >= send_batch_count)
  213. goto over_batch;
  214. spin_lock_irqsave(&cp->cp_lock, flags);
  215. if (!list_empty(&cp->cp_send_queue)) {
  216. rm = list_entry(cp->cp_send_queue.next,
  217. struct rds_message,
  218. m_conn_item);
  219. rds_message_addref(rm);
  220. /*
  221. * Move the message from the send queue to the retransmit
  222. * list right away.
  223. */
  224. list_move_tail(&rm->m_conn_item,
  225. &cp->cp_retrans);
  226. }
  227. spin_unlock_irqrestore(&cp->cp_lock, flags);
  228. if (!rm)
  229. break;
  230. /* Unfortunately, the way Infiniband deals with
  231. * RDMA to a bad MR key is by moving the entire
  232. * queue pair to error state. We cold possibly
  233. * recover from that, but right now we drop the
  234. * connection.
  235. * Therefore, we never retransmit messages with RDMA ops.
  236. */
  237. if (test_bit(RDS_MSG_FLUSH, &rm->m_flags) ||
  238. (rm->rdma.op_active &&
  239. test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags))) {
  240. spin_lock_irqsave(&cp->cp_lock, flags);
  241. if (test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags))
  242. list_move(&rm->m_conn_item, &to_be_dropped);
  243. spin_unlock_irqrestore(&cp->cp_lock, flags);
  244. continue;
  245. }
  246. /* Require an ACK every once in a while */
  247. len = ntohl(rm->m_inc.i_hdr.h_len);
  248. if (cp->cp_unacked_packets == 0 ||
  249. cp->cp_unacked_bytes < len) {
  250. set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
  251. cp->cp_unacked_packets =
  252. rds_sysctl_max_unacked_packets;
  253. cp->cp_unacked_bytes =
  254. rds_sysctl_max_unacked_bytes;
  255. rds_stats_inc(s_send_ack_required);
  256. } else {
  257. cp->cp_unacked_bytes -= len;
  258. cp->cp_unacked_packets--;
  259. }
  260. cp->cp_xmit_rm = rm;
  261. }
  262. /* The transport either sends the whole rdma or none of it */
  263. if (rm->rdma.op_active && !cp->cp_xmit_rdma_sent) {
  264. rm->m_final_op = &rm->rdma;
  265. /* The transport owns the mapped memory for now.
  266. * You can't unmap it while it's on the send queue
  267. */
  268. set_bit(RDS_MSG_MAPPED, &rm->m_flags);
  269. ret = conn->c_trans->xmit_rdma(conn, &rm->rdma);
  270. if (ret) {
  271. clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
  272. wake_up_interruptible(&rm->m_flush_wait);
  273. break;
  274. }
  275. cp->cp_xmit_rdma_sent = 1;
  276. }
  277. if (rm->atomic.op_active && !cp->cp_xmit_atomic_sent) {
  278. rm->m_final_op = &rm->atomic;
  279. /* The transport owns the mapped memory for now.
  280. * You can't unmap it while it's on the send queue
  281. */
  282. set_bit(RDS_MSG_MAPPED, &rm->m_flags);
  283. ret = conn->c_trans->xmit_atomic(conn, &rm->atomic);
  284. if (ret) {
  285. clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
  286. wake_up_interruptible(&rm->m_flush_wait);
  287. break;
  288. }
  289. cp->cp_xmit_atomic_sent = 1;
  290. }
  291. /*
  292. * A number of cases require an RDS header to be sent
  293. * even if there is no data.
  294. * We permit 0-byte sends; rds-ping depends on this.
  295. * However, if there are exclusively attached silent ops,
  296. * we skip the hdr/data send, to enable silent operation.
  297. */
  298. if (rm->data.op_nents == 0) {
  299. int ops_present;
  300. int all_ops_are_silent = 1;
  301. ops_present = (rm->atomic.op_active || rm->rdma.op_active);
  302. if (rm->atomic.op_active && !rm->atomic.op_silent)
  303. all_ops_are_silent = 0;
  304. if (rm->rdma.op_active && !rm->rdma.op_silent)
  305. all_ops_are_silent = 0;
  306. if (ops_present && all_ops_are_silent
  307. && !rm->m_rdma_cookie)
  308. rm->data.op_active = 0;
  309. }
  310. if (rm->data.op_active && !cp->cp_xmit_data_sent) {
  311. rm->m_final_op = &rm->data;
  312. ret = conn->c_trans->xmit(conn, rm,
  313. cp->cp_xmit_hdr_off,
  314. cp->cp_xmit_sg,
  315. cp->cp_xmit_data_off);
  316. if (ret <= 0)
  317. break;
  318. if (cp->cp_xmit_hdr_off < sizeof(struct rds_header)) {
  319. tmp = min_t(int, ret,
  320. sizeof(struct rds_header) -
  321. cp->cp_xmit_hdr_off);
  322. cp->cp_xmit_hdr_off += tmp;
  323. ret -= tmp;
  324. }
  325. sg = &rm->data.op_sg[cp->cp_xmit_sg];
  326. while (ret) {
  327. tmp = min_t(int, ret, sg->length -
  328. cp->cp_xmit_data_off);
  329. cp->cp_xmit_data_off += tmp;
  330. ret -= tmp;
  331. if (cp->cp_xmit_data_off == sg->length) {
  332. cp->cp_xmit_data_off = 0;
  333. sg++;
  334. cp->cp_xmit_sg++;
  335. BUG_ON(ret != 0 && cp->cp_xmit_sg ==
  336. rm->data.op_nents);
  337. }
  338. }
  339. if (cp->cp_xmit_hdr_off == sizeof(struct rds_header) &&
  340. (cp->cp_xmit_sg == rm->data.op_nents))
  341. cp->cp_xmit_data_sent = 1;
  342. }
  343. /*
  344. * A rm will only take multiple times through this loop
  345. * if there is a data op. Thus, if the data is sent (or there was
  346. * none), then we're done with the rm.
  347. */
  348. if (!rm->data.op_active || cp->cp_xmit_data_sent) {
  349. cp->cp_xmit_rm = NULL;
  350. cp->cp_xmit_sg = 0;
  351. cp->cp_xmit_hdr_off = 0;
  352. cp->cp_xmit_data_off = 0;
  353. cp->cp_xmit_rdma_sent = 0;
  354. cp->cp_xmit_atomic_sent = 0;
  355. cp->cp_xmit_data_sent = 0;
  356. rds_message_put(rm);
  357. }
  358. }
  359. over_batch:
  360. if (conn->c_trans->xmit_path_complete)
  361. conn->c_trans->xmit_path_complete(cp);
  362. release_in_xmit(cp);
  363. /* Nuke any messages we decided not to retransmit. */
  364. if (!list_empty(&to_be_dropped)) {
  365. /* irqs on here, so we can put(), unlike above */
  366. list_for_each_entry(rm, &to_be_dropped, m_conn_item)
  367. rds_message_put(rm);
  368. rds_send_remove_from_sock(&to_be_dropped, RDS_RDMA_DROPPED);
  369. }
  370. /*
  371. * Other senders can queue a message after we last test the send queue
  372. * but before we clear RDS_IN_XMIT. In that case they'd back off and
  373. * not try and send their newly queued message. We need to check the
  374. * send queue after having cleared RDS_IN_XMIT so that their message
  375. * doesn't get stuck on the send queue.
  376. *
  377. * If the transport cannot continue (i.e ret != 0), then it must
  378. * call us when more room is available, such as from the tx
  379. * completion handler.
  380. *
  381. * We have an extra generation check here so that if someone manages
  382. * to jump in after our release_in_xmit, we'll see that they have done
  383. * some work and we will skip our goto
  384. */
  385. if (ret == 0) {
  386. bool raced;
  387. smp_mb();
  388. raced = send_gen != READ_ONCE(cp->cp_send_gen);
  389. if ((test_bit(0, &conn->c_map_queued) ||
  390. !list_empty(&cp->cp_send_queue)) && !raced) {
  391. if (batch_count < send_batch_count)
  392. goto restart;
  393. rcu_read_lock();
  394. if (rds_destroy_pending(cp->cp_conn))
  395. ret = -ENETUNREACH;
  396. else
  397. queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
  398. rcu_read_unlock();
  399. } else if (raced) {
  400. rds_stats_inc(s_send_lock_queue_raced);
  401. }
  402. }
  403. out:
  404. return ret;
  405. }
  406. EXPORT_SYMBOL_GPL(rds_send_xmit);
  407. static void rds_send_sndbuf_remove(struct rds_sock *rs, struct rds_message *rm)
  408. {
  409. u32 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
  410. assert_spin_locked(&rs->rs_lock);
  411. BUG_ON(rs->rs_snd_bytes < len);
  412. rs->rs_snd_bytes -= len;
  413. if (rs->rs_snd_bytes == 0)
  414. rds_stats_inc(s_send_queue_empty);
  415. }
  416. static inline int rds_send_is_acked(struct rds_message *rm, u64 ack,
  417. is_acked_func is_acked)
  418. {
  419. if (is_acked)
  420. return is_acked(rm, ack);
  421. return be64_to_cpu(rm->m_inc.i_hdr.h_sequence) <= ack;
  422. }
  423. /*
  424. * This is pretty similar to what happens below in the ACK
  425. * handling code - except that we call here as soon as we get
  426. * the IB send completion on the RDMA op and the accompanying
  427. * message.
  428. */
  429. void rds_rdma_send_complete(struct rds_message *rm, int status)
  430. {
  431. struct rds_sock *rs = NULL;
  432. struct rm_rdma_op *ro;
  433. struct rds_notifier *notifier;
  434. unsigned long flags;
  435. unsigned int notify = 0;
  436. spin_lock_irqsave(&rm->m_rs_lock, flags);
  437. notify = rm->rdma.op_notify | rm->data.op_notify;
  438. ro = &rm->rdma;
  439. if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags) &&
  440. ro->op_active && notify && ro->op_notifier) {
  441. notifier = ro->op_notifier;
  442. rs = rm->m_rs;
  443. sock_hold(rds_rs_to_sk(rs));
  444. notifier->n_status = status;
  445. spin_lock(&rs->rs_lock);
  446. list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
  447. spin_unlock(&rs->rs_lock);
  448. ro->op_notifier = NULL;
  449. }
  450. spin_unlock_irqrestore(&rm->m_rs_lock, flags);
  451. if (rs) {
  452. rds_wake_sk_sleep(rs);
  453. sock_put(rds_rs_to_sk(rs));
  454. }
  455. }
  456. EXPORT_SYMBOL_GPL(rds_rdma_send_complete);
  457. /*
  458. * Just like above, except looks at atomic op
  459. */
  460. void rds_atomic_send_complete(struct rds_message *rm, int status)
  461. {
  462. struct rds_sock *rs = NULL;
  463. struct rm_atomic_op *ao;
  464. struct rds_notifier *notifier;
  465. unsigned long flags;
  466. spin_lock_irqsave(&rm->m_rs_lock, flags);
  467. ao = &rm->atomic;
  468. if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags)
  469. && ao->op_active && ao->op_notify && ao->op_notifier) {
  470. notifier = ao->op_notifier;
  471. rs = rm->m_rs;
  472. sock_hold(rds_rs_to_sk(rs));
  473. notifier->n_status = status;
  474. spin_lock(&rs->rs_lock);
  475. list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
  476. spin_unlock(&rs->rs_lock);
  477. ao->op_notifier = NULL;
  478. }
  479. spin_unlock_irqrestore(&rm->m_rs_lock, flags);
  480. if (rs) {
  481. rds_wake_sk_sleep(rs);
  482. sock_put(rds_rs_to_sk(rs));
  483. }
  484. }
  485. EXPORT_SYMBOL_GPL(rds_atomic_send_complete);
  486. /*
  487. * This is the same as rds_rdma_send_complete except we
  488. * don't do any locking - we have all the ingredients (message,
  489. * socket, socket lock) and can just move the notifier.
  490. */
  491. static inline void
  492. __rds_send_complete(struct rds_sock *rs, struct rds_message *rm, int status)
  493. {
  494. struct rm_rdma_op *ro;
  495. struct rm_atomic_op *ao;
  496. ro = &rm->rdma;
  497. if (ro->op_active && ro->op_notify && ro->op_notifier) {
  498. ro->op_notifier->n_status = status;
  499. list_add_tail(&ro->op_notifier->n_list, &rs->rs_notify_queue);
  500. ro->op_notifier = NULL;
  501. }
  502. ao = &rm->atomic;
  503. if (ao->op_active && ao->op_notify && ao->op_notifier) {
  504. ao->op_notifier->n_status = status;
  505. list_add_tail(&ao->op_notifier->n_list, &rs->rs_notify_queue);
  506. ao->op_notifier = NULL;
  507. }
  508. /* No need to wake the app - caller does this */
  509. }
  510. /*
  511. * This removes messages from the socket's list if they're on it. The list
  512. * argument must be private to the caller, we must be able to modify it
  513. * without locks. The messages must have a reference held for their
  514. * position on the list. This function will drop that reference after
  515. * removing the messages from the 'messages' list regardless of if it found
  516. * the messages on the socket list or not.
  517. */
  518. static void rds_send_remove_from_sock(struct list_head *messages, int status)
  519. {
  520. unsigned long flags;
  521. struct rds_sock *rs = NULL;
  522. struct rds_message *rm;
  523. while (!list_empty(messages)) {
  524. int was_on_sock = 0;
  525. rm = list_entry(messages->next, struct rds_message,
  526. m_conn_item);
  527. list_del_init(&rm->m_conn_item);
  528. /*
  529. * If we see this flag cleared then we're *sure* that someone
  530. * else beat us to removing it from the sock. If we race
  531. * with their flag update we'll get the lock and then really
  532. * see that the flag has been cleared.
  533. *
  534. * The message spinlock makes sure nobody clears rm->m_rs
  535. * while we're messing with it. It does not prevent the
  536. * message from being removed from the socket, though.
  537. */
  538. spin_lock_irqsave(&rm->m_rs_lock, flags);
  539. if (!test_bit(RDS_MSG_ON_SOCK, &rm->m_flags))
  540. goto unlock_and_drop;
  541. if (rs != rm->m_rs) {
  542. if (rs) {
  543. rds_wake_sk_sleep(rs);
  544. sock_put(rds_rs_to_sk(rs));
  545. }
  546. rs = rm->m_rs;
  547. if (rs)
  548. sock_hold(rds_rs_to_sk(rs));
  549. }
  550. if (!rs)
  551. goto unlock_and_drop;
  552. spin_lock(&rs->rs_lock);
  553. if (test_and_clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags)) {
  554. struct rm_rdma_op *ro = &rm->rdma;
  555. struct rds_notifier *notifier;
  556. list_del_init(&rm->m_sock_item);
  557. rds_send_sndbuf_remove(rs, rm);
  558. if (ro->op_active && ro->op_notifier &&
  559. (ro->op_notify || (ro->op_recverr && status))) {
  560. notifier = ro->op_notifier;
  561. list_add_tail(&notifier->n_list,
  562. &rs->rs_notify_queue);
  563. if (!notifier->n_status)
  564. notifier->n_status = status;
  565. rm->rdma.op_notifier = NULL;
  566. }
  567. was_on_sock = 1;
  568. }
  569. spin_unlock(&rs->rs_lock);
  570. unlock_and_drop:
  571. spin_unlock_irqrestore(&rm->m_rs_lock, flags);
  572. rds_message_put(rm);
  573. if (was_on_sock)
  574. rds_message_put(rm);
  575. }
  576. if (rs) {
  577. rds_wake_sk_sleep(rs);
  578. sock_put(rds_rs_to_sk(rs));
  579. }
  580. }
  581. /*
  582. * Transports call here when they've determined that the receiver queued
  583. * messages up to, and including, the given sequence number. Messages are
  584. * moved to the retrans queue when rds_send_xmit picks them off the send
  585. * queue. This means that in the TCP case, the message may not have been
  586. * assigned the m_ack_seq yet - but that's fine as long as tcp_is_acked
  587. * checks the RDS_MSG_HAS_ACK_SEQ bit.
  588. */
  589. void rds_send_path_drop_acked(struct rds_conn_path *cp, u64 ack,
  590. is_acked_func is_acked)
  591. {
  592. struct rds_message *rm, *tmp;
  593. unsigned long flags;
  594. LIST_HEAD(list);
  595. spin_lock_irqsave(&cp->cp_lock, flags);
  596. list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
  597. if (!rds_send_is_acked(rm, ack, is_acked))
  598. break;
  599. list_move(&rm->m_conn_item, &list);
  600. clear_bit(RDS_MSG_ON_CONN, &rm->m_flags);
  601. }
  602. /* order flag updates with spin locks */
  603. if (!list_empty(&list))
  604. smp_mb__after_atomic();
  605. spin_unlock_irqrestore(&cp->cp_lock, flags);
  606. /* now remove the messages from the sock list as needed */
  607. rds_send_remove_from_sock(&list, RDS_RDMA_SUCCESS);
  608. }
  609. EXPORT_SYMBOL_GPL(rds_send_path_drop_acked);
  610. void rds_send_drop_acked(struct rds_connection *conn, u64 ack,
  611. is_acked_func is_acked)
  612. {
  613. WARN_ON(conn->c_trans->t_mp_capable);
  614. rds_send_path_drop_acked(&conn->c_path[0], ack, is_acked);
  615. }
  616. EXPORT_SYMBOL_GPL(rds_send_drop_acked);
  617. void rds_send_drop_to(struct rds_sock *rs, struct sockaddr_in6 *dest)
  618. {
  619. struct rds_message *rm, *tmp;
  620. struct rds_connection *conn;
  621. struct rds_conn_path *cp;
  622. unsigned long flags;
  623. LIST_HEAD(list);
  624. /* get all the messages we're dropping under the rs lock */
  625. spin_lock_irqsave(&rs->rs_lock, flags);
  626. list_for_each_entry_safe(rm, tmp, &rs->rs_send_queue, m_sock_item) {
  627. if (dest &&
  628. (!ipv6_addr_equal(&dest->sin6_addr, &rm->m_daddr) ||
  629. dest->sin6_port != rm->m_inc.i_hdr.h_dport))
  630. continue;
  631. list_move(&rm->m_sock_item, &list);
  632. rds_send_sndbuf_remove(rs, rm);
  633. clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
  634. }
  635. /* order flag updates with the rs lock */
  636. smp_mb__after_atomic();
  637. spin_unlock_irqrestore(&rs->rs_lock, flags);
  638. if (list_empty(&list))
  639. return;
  640. /* Remove the messages from the conn */
  641. list_for_each_entry(rm, &list, m_sock_item) {
  642. conn = rm->m_inc.i_conn;
  643. if (conn->c_trans->t_mp_capable)
  644. cp = rm->m_inc.i_conn_path;
  645. else
  646. cp = &conn->c_path[0];
  647. spin_lock_irqsave(&cp->cp_lock, flags);
  648. /*
  649. * Maybe someone else beat us to removing rm from the conn.
  650. * If we race with their flag update we'll get the lock and
  651. * then really see that the flag has been cleared.
  652. */
  653. if (!test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags)) {
  654. spin_unlock_irqrestore(&cp->cp_lock, flags);
  655. continue;
  656. }
  657. list_del_init(&rm->m_conn_item);
  658. spin_unlock_irqrestore(&cp->cp_lock, flags);
  659. /*
  660. * Couldn't grab m_rs_lock in top loop (lock ordering),
  661. * but we can now.
  662. */
  663. spin_lock_irqsave(&rm->m_rs_lock, flags);
  664. spin_lock(&rs->rs_lock);
  665. __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
  666. spin_unlock(&rs->rs_lock);
  667. spin_unlock_irqrestore(&rm->m_rs_lock, flags);
  668. rds_message_put(rm);
  669. }
  670. rds_wake_sk_sleep(rs);
  671. while (!list_empty(&list)) {
  672. rm = list_entry(list.next, struct rds_message, m_sock_item);
  673. list_del_init(&rm->m_sock_item);
  674. rds_message_wait(rm);
  675. /* just in case the code above skipped this message
  676. * because RDS_MSG_ON_CONN wasn't set, run it again here
  677. * taking m_rs_lock is the only thing that keeps us
  678. * from racing with ack processing.
  679. */
  680. spin_lock_irqsave(&rm->m_rs_lock, flags);
  681. spin_lock(&rs->rs_lock);
  682. __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
  683. spin_unlock(&rs->rs_lock);
  684. spin_unlock_irqrestore(&rm->m_rs_lock, flags);
  685. rds_message_put(rm);
  686. }
  687. }
  688. /*
  689. * we only want this to fire once so we use the callers 'queued'. It's
  690. * possible that another thread can race with us and remove the
  691. * message from the flow with RDS_CANCEL_SENT_TO.
  692. */
  693. static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
  694. struct rds_conn_path *cp,
  695. struct rds_message *rm, __be16 sport,
  696. __be16 dport, int *queued)
  697. {
  698. unsigned long flags;
  699. u32 len;
  700. if (*queued)
  701. goto out;
  702. len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
  703. /* this is the only place which holds both the socket's rs_lock
  704. * and the connection's c_lock */
  705. spin_lock_irqsave(&rs->rs_lock, flags);
  706. /*
  707. * If there is a little space in sndbuf, we don't queue anything,
  708. * and userspace gets -EAGAIN. But poll() indicates there's send
  709. * room. This can lead to bad behavior (spinning) if snd_bytes isn't
  710. * freed up by incoming acks. So we check the *old* value of
  711. * rs_snd_bytes here to allow the last msg to exceed the buffer,
  712. * and poll() now knows no more data can be sent.
  713. */
  714. if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
  715. rs->rs_snd_bytes += len;
  716. /* let recv side know we are close to send space exhaustion.
  717. * This is probably not the optimal way to do it, as this
  718. * means we set the flag on *all* messages as soon as our
  719. * throughput hits a certain threshold.
  720. */
  721. if (rs->rs_snd_bytes >= rds_sk_sndbuf(rs) / 2)
  722. set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
  723. list_add_tail(&rm->m_sock_item, &rs->rs_send_queue);
  724. set_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
  725. rds_message_addref(rm);
  726. sock_hold(rds_rs_to_sk(rs));
  727. rm->m_rs = rs;
  728. /* The code ordering is a little weird, but we're
  729. trying to minimize the time we hold c_lock */
  730. rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
  731. rm->m_inc.i_conn = conn;
  732. rm->m_inc.i_conn_path = cp;
  733. rds_message_addref(rm);
  734. spin_lock(&cp->cp_lock);
  735. rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
  736. list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
  737. set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
  738. spin_unlock(&cp->cp_lock);
  739. rdsdebug("queued msg %p len %d, rs %p bytes %d seq %llu\n",
  740. rm, len, rs, rs->rs_snd_bytes,
  741. (unsigned long long)be64_to_cpu(rm->m_inc.i_hdr.h_sequence));
  742. *queued = 1;
  743. }
  744. spin_unlock_irqrestore(&rs->rs_lock, flags);
  745. out:
  746. return *queued;
  747. }
  748. /*
  749. * rds_message is getting to be quite complicated, and we'd like to allocate
  750. * it all in one go. This figures out how big it needs to be up front.
  751. */
  752. static int rds_rm_size(struct msghdr *msg, int num_sgs)
  753. {
  754. struct cmsghdr *cmsg;
  755. int size = 0;
  756. int cmsg_groups = 0;
  757. int retval;
  758. bool zcopy_cookie = false;
  759. for_each_cmsghdr(cmsg, msg) {
  760. if (!CMSG_OK(msg, cmsg))
  761. return -EINVAL;
  762. if (cmsg->cmsg_level != SOL_RDS)
  763. continue;
  764. switch (cmsg->cmsg_type) {
  765. case RDS_CMSG_RDMA_ARGS:
  766. cmsg_groups |= 1;
  767. retval = rds_rdma_extra_size(CMSG_DATA(cmsg));
  768. if (retval < 0)
  769. return retval;
  770. size += retval;
  771. break;
  772. case RDS_CMSG_ZCOPY_COOKIE:
  773. zcopy_cookie = true;
  774. /* fall through */
  775. case RDS_CMSG_RDMA_DEST:
  776. case RDS_CMSG_RDMA_MAP:
  777. cmsg_groups |= 2;
  778. /* these are valid but do no add any size */
  779. break;
  780. case RDS_CMSG_ATOMIC_CSWP:
  781. case RDS_CMSG_ATOMIC_FADD:
  782. case RDS_CMSG_MASKED_ATOMIC_CSWP:
  783. case RDS_CMSG_MASKED_ATOMIC_FADD:
  784. cmsg_groups |= 1;
  785. size += sizeof(struct scatterlist);
  786. break;
  787. default:
  788. return -EINVAL;
  789. }
  790. }
  791. if ((msg->msg_flags & MSG_ZEROCOPY) && !zcopy_cookie)
  792. return -EINVAL;
  793. size += num_sgs * sizeof(struct scatterlist);
  794. /* Ensure (DEST, MAP) are never used with (ARGS, ATOMIC) */
  795. if (cmsg_groups == 3)
  796. return -EINVAL;
  797. return size;
  798. }
  799. static int rds_cmsg_zcopy(struct rds_sock *rs, struct rds_message *rm,
  800. struct cmsghdr *cmsg)
  801. {
  802. u32 *cookie;
  803. if (cmsg->cmsg_len < CMSG_LEN(sizeof(*cookie)) ||
  804. !rm->data.op_mmp_znotifier)
  805. return -EINVAL;
  806. cookie = CMSG_DATA(cmsg);
  807. rm->data.op_mmp_znotifier->z_cookie = *cookie;
  808. return 0;
  809. }
  810. static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
  811. struct msghdr *msg, int *allocated_mr)
  812. {
  813. struct cmsghdr *cmsg;
  814. int ret = 0;
  815. for_each_cmsghdr(cmsg, msg) {
  816. if (!CMSG_OK(msg, cmsg))
  817. return -EINVAL;
  818. if (cmsg->cmsg_level != SOL_RDS)
  819. continue;
  820. /* As a side effect, RDMA_DEST and RDMA_MAP will set
  821. * rm->rdma.m_rdma_cookie and rm->rdma.m_rdma_mr.
  822. */
  823. switch (cmsg->cmsg_type) {
  824. case RDS_CMSG_RDMA_ARGS:
  825. ret = rds_cmsg_rdma_args(rs, rm, cmsg);
  826. break;
  827. case RDS_CMSG_RDMA_DEST:
  828. ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
  829. break;
  830. case RDS_CMSG_RDMA_MAP:
  831. ret = rds_cmsg_rdma_map(rs, rm, cmsg);
  832. if (!ret)
  833. *allocated_mr = 1;
  834. else if (ret == -ENODEV)
  835. /* Accommodate the get_mr() case which can fail
  836. * if connection isn't established yet.
  837. */
  838. ret = -EAGAIN;
  839. break;
  840. case RDS_CMSG_ATOMIC_CSWP:
  841. case RDS_CMSG_ATOMIC_FADD:
  842. case RDS_CMSG_MASKED_ATOMIC_CSWP:
  843. case RDS_CMSG_MASKED_ATOMIC_FADD:
  844. ret = rds_cmsg_atomic(rs, rm, cmsg);
  845. break;
  846. case RDS_CMSG_ZCOPY_COOKIE:
  847. ret = rds_cmsg_zcopy(rs, rm, cmsg);
  848. break;
  849. default:
  850. return -EINVAL;
  851. }
  852. if (ret)
  853. break;
  854. }
  855. return ret;
  856. }
  857. static int rds_send_mprds_hash(struct rds_sock *rs,
  858. struct rds_connection *conn, int nonblock)
  859. {
  860. int hash;
  861. if (conn->c_npaths == 0)
  862. hash = RDS_MPATH_HASH(rs, RDS_MPATH_WORKERS);
  863. else
  864. hash = RDS_MPATH_HASH(rs, conn->c_npaths);
  865. if (conn->c_npaths == 0 && hash != 0) {
  866. rds_send_ping(conn, 0);
  867. /* The underlying connection is not up yet. Need to wait
  868. * until it is up to be sure that the non-zero c_path can be
  869. * used. But if we are interrupted, we have to use the zero
  870. * c_path in case the connection ends up being non-MP capable.
  871. */
  872. if (conn->c_npaths == 0) {
  873. /* Cannot wait for the connection be made, so just use
  874. * the base c_path.
  875. */
  876. if (nonblock)
  877. return 0;
  878. if (wait_event_interruptible(conn->c_hs_waitq,
  879. conn->c_npaths != 0))
  880. hash = 0;
  881. }
  882. if (conn->c_npaths == 1)
  883. hash = 0;
  884. }
  885. return hash;
  886. }
  887. static int rds_rdma_bytes(struct msghdr *msg, size_t *rdma_bytes)
  888. {
  889. struct rds_rdma_args *args;
  890. struct cmsghdr *cmsg;
  891. for_each_cmsghdr(cmsg, msg) {
  892. if (!CMSG_OK(msg, cmsg))
  893. return -EINVAL;
  894. if (cmsg->cmsg_level != SOL_RDS)
  895. continue;
  896. if (cmsg->cmsg_type == RDS_CMSG_RDMA_ARGS) {
  897. if (cmsg->cmsg_len <
  898. CMSG_LEN(sizeof(struct rds_rdma_args)))
  899. return -EINVAL;
  900. args = CMSG_DATA(cmsg);
  901. *rdma_bytes += args->remote_vec.bytes;
  902. }
  903. }
  904. return 0;
  905. }
  906. int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
  907. {
  908. struct sock *sk = sock->sk;
  909. struct rds_sock *rs = rds_sk_to_rs(sk);
  910. DECLARE_SOCKADDR(struct sockaddr_in6 *, sin6, msg->msg_name);
  911. DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
  912. __be16 dport;
  913. struct rds_message *rm = NULL;
  914. struct rds_connection *conn;
  915. int ret = 0;
  916. int queued = 0, allocated_mr = 0;
  917. int nonblock = msg->msg_flags & MSG_DONTWAIT;
  918. long timeo = sock_sndtimeo(sk, nonblock);
  919. struct rds_conn_path *cpath;
  920. struct in6_addr daddr;
  921. __u32 scope_id = 0;
  922. size_t total_payload_len = payload_len, rdma_payload_len = 0;
  923. bool zcopy = ((msg->msg_flags & MSG_ZEROCOPY) &&
  924. sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY));
  925. int num_sgs = ceil(payload_len, PAGE_SIZE);
  926. int namelen;
  927. /* Mirror Linux UDP mirror of BSD error message compatibility */
  928. /* XXX: Perhaps MSG_MORE someday */
  929. if (msg->msg_flags & ~(MSG_DONTWAIT | MSG_CMSG_COMPAT | MSG_ZEROCOPY)) {
  930. ret = -EOPNOTSUPP;
  931. goto out;
  932. }
  933. namelen = msg->msg_namelen;
  934. if (namelen != 0) {
  935. if (namelen < sizeof(*usin)) {
  936. ret = -EINVAL;
  937. goto out;
  938. }
  939. switch (usin->sin_family) {
  940. case AF_INET:
  941. if (usin->sin_addr.s_addr == htonl(INADDR_ANY) ||
  942. usin->sin_addr.s_addr == htonl(INADDR_BROADCAST) ||
  943. IN_MULTICAST(ntohl(usin->sin_addr.s_addr))) {
  944. ret = -EINVAL;
  945. goto out;
  946. }
  947. ipv6_addr_set_v4mapped(usin->sin_addr.s_addr, &daddr);
  948. dport = usin->sin_port;
  949. break;
  950. #if IS_ENABLED(CONFIG_IPV6)
  951. case AF_INET6: {
  952. int addr_type;
  953. if (namelen < sizeof(*sin6)) {
  954. ret = -EINVAL;
  955. goto out;
  956. }
  957. addr_type = ipv6_addr_type(&sin6->sin6_addr);
  958. if (!(addr_type & IPV6_ADDR_UNICAST)) {
  959. __be32 addr4;
  960. if (!(addr_type & IPV6_ADDR_MAPPED)) {
  961. ret = -EINVAL;
  962. goto out;
  963. }
  964. /* It is a mapped address. Need to do some
  965. * sanity checks.
  966. */
  967. addr4 = sin6->sin6_addr.s6_addr32[3];
  968. if (addr4 == htonl(INADDR_ANY) ||
  969. addr4 == htonl(INADDR_BROADCAST) ||
  970. IN_MULTICAST(ntohl(addr4))) {
  971. ret = -EINVAL;
  972. goto out;
  973. }
  974. }
  975. if (addr_type & IPV6_ADDR_LINKLOCAL) {
  976. if (sin6->sin6_scope_id == 0) {
  977. ret = -EINVAL;
  978. goto out;
  979. }
  980. scope_id = sin6->sin6_scope_id;
  981. }
  982. daddr = sin6->sin6_addr;
  983. dport = sin6->sin6_port;
  984. break;
  985. }
  986. #endif
  987. default:
  988. ret = -EINVAL;
  989. goto out;
  990. }
  991. } else {
  992. /* We only care about consistency with ->connect() */
  993. lock_sock(sk);
  994. daddr = rs->rs_conn_addr;
  995. dport = rs->rs_conn_port;
  996. scope_id = rs->rs_bound_scope_id;
  997. release_sock(sk);
  998. }
  999. lock_sock(sk);
  1000. if (ipv6_addr_any(&rs->rs_bound_addr) || ipv6_addr_any(&daddr)) {
  1001. release_sock(sk);
  1002. ret = -ENOTCONN;
  1003. goto out;
  1004. } else if (namelen != 0) {
  1005. /* Cannot send to an IPv4 address using an IPv6 source
  1006. * address and cannot send to an IPv6 address using an
  1007. * IPv4 source address.
  1008. */
  1009. if (ipv6_addr_v4mapped(&daddr) ^
  1010. ipv6_addr_v4mapped(&rs->rs_bound_addr)) {
  1011. release_sock(sk);
  1012. ret = -EOPNOTSUPP;
  1013. goto out;
  1014. }
  1015. /* If the socket is already bound to a link local address,
  1016. * it can only send to peers on the same link. But allow
  1017. * communicating beween link local and non-link local address.
  1018. */
  1019. if (scope_id != rs->rs_bound_scope_id) {
  1020. if (!scope_id) {
  1021. scope_id = rs->rs_bound_scope_id;
  1022. } else if (rs->rs_bound_scope_id) {
  1023. release_sock(sk);
  1024. ret = -EINVAL;
  1025. goto out;
  1026. }
  1027. }
  1028. }
  1029. release_sock(sk);
  1030. ret = rds_rdma_bytes(msg, &rdma_payload_len);
  1031. if (ret)
  1032. goto out;
  1033. total_payload_len += rdma_payload_len;
  1034. if (max_t(size_t, payload_len, rdma_payload_len) > RDS_MAX_MSG_SIZE) {
  1035. ret = -EMSGSIZE;
  1036. goto out;
  1037. }
  1038. if (payload_len > rds_sk_sndbuf(rs)) {
  1039. ret = -EMSGSIZE;
  1040. goto out;
  1041. }
  1042. if (zcopy) {
  1043. if (rs->rs_transport->t_type != RDS_TRANS_TCP) {
  1044. ret = -EOPNOTSUPP;
  1045. goto out;
  1046. }
  1047. num_sgs = iov_iter_npages(&msg->msg_iter, INT_MAX);
  1048. }
  1049. /* size of rm including all sgs */
  1050. ret = rds_rm_size(msg, num_sgs);
  1051. if (ret < 0)
  1052. goto out;
  1053. rm = rds_message_alloc(ret, GFP_KERNEL);
  1054. if (!rm) {
  1055. ret = -ENOMEM;
  1056. goto out;
  1057. }
  1058. /* Attach data to the rm */
  1059. if (payload_len) {
  1060. rm->data.op_sg = rds_message_alloc_sgs(rm, num_sgs);
  1061. if (!rm->data.op_sg) {
  1062. ret = -ENOMEM;
  1063. goto out;
  1064. }
  1065. ret = rds_message_copy_from_user(rm, &msg->msg_iter, zcopy);
  1066. if (ret)
  1067. goto out;
  1068. }
  1069. rm->data.op_active = 1;
  1070. rm->m_daddr = daddr;
  1071. /* rds_conn_create has a spinlock that runs with IRQ off.
  1072. * Caching the conn in the socket helps a lot. */
  1073. if (rs->rs_conn && ipv6_addr_equal(&rs->rs_conn->c_faddr, &daddr))
  1074. conn = rs->rs_conn;
  1075. else {
  1076. conn = rds_conn_create_outgoing(sock_net(sock->sk),
  1077. &rs->rs_bound_addr, &daddr,
  1078. rs->rs_transport,
  1079. sock->sk->sk_allocation,
  1080. scope_id);
  1081. if (IS_ERR(conn)) {
  1082. ret = PTR_ERR(conn);
  1083. goto out;
  1084. }
  1085. rs->rs_conn = conn;
  1086. }
  1087. if (conn->c_trans->t_mp_capable)
  1088. cpath = &conn->c_path[rds_send_mprds_hash(rs, conn, nonblock)];
  1089. else
  1090. cpath = &conn->c_path[0];
  1091. rm->m_conn_path = cpath;
  1092. /* Parse any control messages the user may have included. */
  1093. ret = rds_cmsg_send(rs, rm, msg, &allocated_mr);
  1094. if (ret) {
  1095. /* Trigger connection so that its ready for the next retry */
  1096. if (ret == -EAGAIN)
  1097. rds_conn_connect_if_down(conn);
  1098. goto out;
  1099. }
  1100. if (rm->rdma.op_active && !conn->c_trans->xmit_rdma) {
  1101. printk_ratelimited(KERN_NOTICE "rdma_op %p conn xmit_rdma %p\n",
  1102. &rm->rdma, conn->c_trans->xmit_rdma);
  1103. ret = -EOPNOTSUPP;
  1104. goto out;
  1105. }
  1106. if (rm->atomic.op_active && !conn->c_trans->xmit_atomic) {
  1107. printk_ratelimited(KERN_NOTICE "atomic_op %p conn xmit_atomic %p\n",
  1108. &rm->atomic, conn->c_trans->xmit_atomic);
  1109. ret = -EOPNOTSUPP;
  1110. goto out;
  1111. }
  1112. if (rds_destroy_pending(conn)) {
  1113. ret = -EAGAIN;
  1114. goto out;
  1115. }
  1116. rds_conn_path_connect_if_down(cpath);
  1117. ret = rds_cong_wait(conn->c_fcong, dport, nonblock, rs);
  1118. if (ret) {
  1119. rs->rs_seen_congestion = 1;
  1120. goto out;
  1121. }
  1122. while (!rds_send_queue_rm(rs, conn, cpath, rm, rs->rs_bound_port,
  1123. dport, &queued)) {
  1124. rds_stats_inc(s_send_queue_full);
  1125. if (nonblock) {
  1126. ret = -EAGAIN;
  1127. goto out;
  1128. }
  1129. timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
  1130. rds_send_queue_rm(rs, conn, cpath, rm,
  1131. rs->rs_bound_port,
  1132. dport,
  1133. &queued),
  1134. timeo);
  1135. rdsdebug("sendmsg woke queued %d timeo %ld\n", queued, timeo);
  1136. if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
  1137. continue;
  1138. ret = timeo;
  1139. if (ret == 0)
  1140. ret = -ETIMEDOUT;
  1141. goto out;
  1142. }
  1143. /*
  1144. * By now we've committed to the send. We reuse rds_send_worker()
  1145. * to retry sends in the rds thread if the transport asks us to.
  1146. */
  1147. rds_stats_inc(s_send_queued);
  1148. ret = rds_send_xmit(cpath);
  1149. if (ret == -ENOMEM || ret == -EAGAIN) {
  1150. ret = 0;
  1151. rcu_read_lock();
  1152. if (rds_destroy_pending(cpath->cp_conn))
  1153. ret = -ENETUNREACH;
  1154. else
  1155. queue_delayed_work(rds_wq, &cpath->cp_send_w, 1);
  1156. rcu_read_unlock();
  1157. }
  1158. if (ret)
  1159. goto out;
  1160. rds_message_put(rm);
  1161. return payload_len;
  1162. out:
  1163. /* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
  1164. * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
  1165. * or in any other way, we need to destroy the MR again */
  1166. if (allocated_mr)
  1167. rds_rdma_unuse(rs, rds_rdma_cookie_key(rm->m_rdma_cookie), 1);
  1168. if (rm)
  1169. rds_message_put(rm);
  1170. return ret;
  1171. }
  1172. /*
  1173. * send out a probe. Can be shared by rds_send_ping,
  1174. * rds_send_pong, rds_send_hb.
  1175. * rds_send_hb should use h_flags
  1176. * RDS_FLAG_HB_PING|RDS_FLAG_ACK_REQUIRED
  1177. * or
  1178. * RDS_FLAG_HB_PONG|RDS_FLAG_ACK_REQUIRED
  1179. */
  1180. static int
  1181. rds_send_probe(struct rds_conn_path *cp, __be16 sport,
  1182. __be16 dport, u8 h_flags)
  1183. {
  1184. struct rds_message *rm;
  1185. unsigned long flags;
  1186. int ret = 0;
  1187. rm = rds_message_alloc(0, GFP_ATOMIC);
  1188. if (!rm) {
  1189. ret = -ENOMEM;
  1190. goto out;
  1191. }
  1192. rm->m_daddr = cp->cp_conn->c_faddr;
  1193. rm->data.op_active = 1;
  1194. rds_conn_path_connect_if_down(cp);
  1195. ret = rds_cong_wait(cp->cp_conn->c_fcong, dport, 1, NULL);
  1196. if (ret)
  1197. goto out;
  1198. spin_lock_irqsave(&cp->cp_lock, flags);
  1199. list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
  1200. set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
  1201. rds_message_addref(rm);
  1202. rm->m_inc.i_conn = cp->cp_conn;
  1203. rm->m_inc.i_conn_path = cp;
  1204. rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport,
  1205. cp->cp_next_tx_seq);
  1206. rm->m_inc.i_hdr.h_flags |= h_flags;
  1207. cp->cp_next_tx_seq++;
  1208. if (RDS_HS_PROBE(be16_to_cpu(sport), be16_to_cpu(dport)) &&
  1209. cp->cp_conn->c_trans->t_mp_capable) {
  1210. u16 npaths = cpu_to_be16(RDS_MPATH_WORKERS);
  1211. u32 my_gen_num = cpu_to_be32(cp->cp_conn->c_my_gen_num);
  1212. rds_message_add_extension(&rm->m_inc.i_hdr,
  1213. RDS_EXTHDR_NPATHS, &npaths,
  1214. sizeof(npaths));
  1215. rds_message_add_extension(&rm->m_inc.i_hdr,
  1216. RDS_EXTHDR_GEN_NUM,
  1217. &my_gen_num,
  1218. sizeof(u32));
  1219. }
  1220. spin_unlock_irqrestore(&cp->cp_lock, flags);
  1221. rds_stats_inc(s_send_queued);
  1222. rds_stats_inc(s_send_pong);
  1223. /* schedule the send work on rds_wq */
  1224. rcu_read_lock();
  1225. if (!rds_destroy_pending(cp->cp_conn))
  1226. queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
  1227. rcu_read_unlock();
  1228. rds_message_put(rm);
  1229. return 0;
  1230. out:
  1231. if (rm)
  1232. rds_message_put(rm);
  1233. return ret;
  1234. }
  1235. int
  1236. rds_send_pong(struct rds_conn_path *cp, __be16 dport)
  1237. {
  1238. return rds_send_probe(cp, 0, dport, 0);
  1239. }
  1240. void
  1241. rds_send_ping(struct rds_connection *conn, int cp_index)
  1242. {
  1243. unsigned long flags;
  1244. struct rds_conn_path *cp = &conn->c_path[cp_index];
  1245. spin_lock_irqsave(&cp->cp_lock, flags);
  1246. if (conn->c_ping_triggered) {
  1247. spin_unlock_irqrestore(&cp->cp_lock, flags);
  1248. return;
  1249. }
  1250. conn->c_ping_triggered = 1;
  1251. spin_unlock_irqrestore(&cp->cp_lock, flags);
  1252. rds_send_probe(cp, cpu_to_be16(RDS_FLAG_PROBE_PORT), 0, 0);
  1253. }
  1254. EXPORT_SYMBOL_GPL(rds_send_ping);