send.c 39 KB

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