ib_send.c 30 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018
  1. /*
  2. * Copyright (c) 2006, 2019 Oracle and/or its affiliates. All rights reserved.
  3. *
  4. * This software is available to you under a choice of one of two
  5. * licenses. You may choose to be licensed under the terms of the GNU
  6. * General Public License (GPL) Version 2, available from the file
  7. * COPYING in the main directory of this source tree, or the
  8. * OpenIB.org BSD license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or
  11. * without modification, are permitted provided that the following
  12. * conditions are met:
  13. *
  14. * - Redistributions of source code must retain the above
  15. * copyright notice, this list of conditions and the following
  16. * disclaimer.
  17. *
  18. * - Redistributions in binary form must reproduce the above
  19. * copyright notice, this list of conditions and the following
  20. * disclaimer in the documentation and/or other materials
  21. * provided with the distribution.
  22. *
  23. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  24. * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  25. * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  26. * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  27. * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  28. * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  29. * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  30. * SOFTWARE.
  31. *
  32. */
  33. #include <linux/kernel.h>
  34. #include <linux/in.h>
  35. #include <linux/device.h>
  36. #include <linux/dmapool.h>
  37. #include <linux/ratelimit.h>
  38. #include "rds_single_path.h"
  39. #include "rds.h"
  40. #include "ib.h"
  41. #include "ib_mr.h"
  42. /*
  43. * Convert IB-specific error message to RDS error message and call core
  44. * completion handler.
  45. */
  46. static void rds_ib_send_complete(struct rds_message *rm,
  47. int wc_status,
  48. void (*complete)(struct rds_message *rm, int status))
  49. {
  50. int notify_status;
  51. switch (wc_status) {
  52. case IB_WC_WR_FLUSH_ERR:
  53. return;
  54. case IB_WC_SUCCESS:
  55. notify_status = RDS_RDMA_SUCCESS;
  56. break;
  57. case IB_WC_REM_ACCESS_ERR:
  58. notify_status = RDS_RDMA_REMOTE_ERROR;
  59. break;
  60. default:
  61. notify_status = RDS_RDMA_OTHER_ERROR;
  62. break;
  63. }
  64. complete(rm, notify_status);
  65. }
  66. static void rds_ib_send_unmap_data(struct rds_ib_connection *ic,
  67. struct rm_data_op *op,
  68. int wc_status)
  69. {
  70. if (op->op_nents)
  71. ib_dma_unmap_sg(ic->i_cm_id->device,
  72. op->op_sg, op->op_nents,
  73. DMA_TO_DEVICE);
  74. }
  75. static void rds_ib_send_unmap_rdma(struct rds_ib_connection *ic,
  76. struct rm_rdma_op *op,
  77. int wc_status)
  78. {
  79. if (op->op_mapped) {
  80. ib_dma_unmap_sg(ic->i_cm_id->device,
  81. op->op_sg, op->op_nents,
  82. op->op_write ? DMA_TO_DEVICE : DMA_FROM_DEVICE);
  83. op->op_mapped = 0;
  84. }
  85. /* If the user asked for a completion notification on this
  86. * message, we can implement three different semantics:
  87. * 1. Notify when we received the ACK on the RDS message
  88. * that was queued with the RDMA. This provides reliable
  89. * notification of RDMA status at the expense of a one-way
  90. * packet delay.
  91. * 2. Notify when the IB stack gives us the completion event for
  92. * the RDMA operation.
  93. * 3. Notify when the IB stack gives us the completion event for
  94. * the accompanying RDS messages.
  95. * Here, we implement approach #3. To implement approach #2,
  96. * we would need to take an event for the rdma WR. To implement #1,
  97. * don't call rds_rdma_send_complete at all, and fall back to the notify
  98. * handling in the ACK processing code.
  99. *
  100. * Note: There's no need to explicitly sync any RDMA buffers using
  101. * ib_dma_sync_sg_for_cpu - the completion for the RDMA
  102. * operation itself unmapped the RDMA buffers, which takes care
  103. * of synching.
  104. */
  105. rds_ib_send_complete(container_of(op, struct rds_message, rdma),
  106. wc_status, rds_rdma_send_complete);
  107. if (op->op_write)
  108. rds_stats_add(s_send_rdma_bytes, op->op_bytes);
  109. else
  110. rds_stats_add(s_recv_rdma_bytes, op->op_bytes);
  111. }
  112. static void rds_ib_send_unmap_atomic(struct rds_ib_connection *ic,
  113. struct rm_atomic_op *op,
  114. int wc_status)
  115. {
  116. /* unmap atomic recvbuf */
  117. if (op->op_mapped) {
  118. ib_dma_unmap_sg(ic->i_cm_id->device, op->op_sg, 1,
  119. DMA_FROM_DEVICE);
  120. op->op_mapped = 0;
  121. }
  122. rds_ib_send_complete(container_of(op, struct rds_message, atomic),
  123. wc_status, rds_atomic_send_complete);
  124. if (op->op_type == RDS_ATOMIC_TYPE_CSWP)
  125. rds_ib_stats_inc(s_ib_atomic_cswp);
  126. else
  127. rds_ib_stats_inc(s_ib_atomic_fadd);
  128. }
  129. /*
  130. * Unmap the resources associated with a struct send_work.
  131. *
  132. * Returns the rm for no good reason other than it is unobtainable
  133. * other than by switching on wr.opcode, currently, and the caller,
  134. * the event handler, needs it.
  135. */
  136. static struct rds_message *rds_ib_send_unmap_op(struct rds_ib_connection *ic,
  137. struct rds_ib_send_work *send,
  138. int wc_status)
  139. {
  140. struct rds_message *rm = NULL;
  141. /* In the error case, wc.opcode sometimes contains garbage */
  142. switch (send->s_wr.opcode) {
  143. case IB_WR_SEND:
  144. if (send->s_op) {
  145. rm = container_of(send->s_op, struct rds_message, data);
  146. rds_ib_send_unmap_data(ic, send->s_op, wc_status);
  147. }
  148. break;
  149. case IB_WR_RDMA_WRITE:
  150. case IB_WR_RDMA_READ:
  151. if (send->s_op) {
  152. rm = container_of(send->s_op, struct rds_message, rdma);
  153. rds_ib_send_unmap_rdma(ic, send->s_op, wc_status);
  154. }
  155. break;
  156. case IB_WR_ATOMIC_FETCH_AND_ADD:
  157. case IB_WR_ATOMIC_CMP_AND_SWP:
  158. if (send->s_op) {
  159. rm = container_of(send->s_op, struct rds_message, atomic);
  160. rds_ib_send_unmap_atomic(ic, send->s_op, wc_status);
  161. }
  162. break;
  163. default:
  164. printk_ratelimited(KERN_NOTICE
  165. "RDS/IB: %s: unexpected opcode 0x%x in WR!\n",
  166. __func__, send->s_wr.opcode);
  167. break;
  168. }
  169. send->s_wr.opcode = 0xdead;
  170. return rm;
  171. }
  172. void rds_ib_send_init_ring(struct rds_ib_connection *ic)
  173. {
  174. struct rds_ib_send_work *send;
  175. u32 i;
  176. for (i = 0, send = ic->i_sends; i < ic->i_send_ring.w_nr; i++, send++) {
  177. struct ib_sge *sge;
  178. send->s_op = NULL;
  179. send->s_wr.wr_id = i;
  180. send->s_wr.sg_list = send->s_sge;
  181. send->s_wr.ex.imm_data = 0;
  182. sge = &send->s_sge[0];
  183. sge->addr = ic->i_send_hdrs_dma[i];
  184. sge->length = sizeof(struct rds_header);
  185. sge->lkey = ic->i_pd->local_dma_lkey;
  186. send->s_sge[1].lkey = ic->i_pd->local_dma_lkey;
  187. }
  188. }
  189. void rds_ib_send_clear_ring(struct rds_ib_connection *ic)
  190. {
  191. struct rds_ib_send_work *send;
  192. u32 i;
  193. for (i = 0, send = ic->i_sends; i < ic->i_send_ring.w_nr; i++, send++) {
  194. if (send->s_op && send->s_wr.opcode != 0xdead)
  195. rds_ib_send_unmap_op(ic, send, IB_WC_WR_FLUSH_ERR);
  196. }
  197. }
  198. /*
  199. * The only fast path caller always has a non-zero nr, so we don't
  200. * bother testing nr before performing the atomic sub.
  201. */
  202. static void rds_ib_sub_signaled(struct rds_ib_connection *ic, int nr)
  203. {
  204. if ((atomic_sub_return(nr, &ic->i_signaled_sends) == 0) &&
  205. waitqueue_active(&rds_ib_ring_empty_wait))
  206. wake_up(&rds_ib_ring_empty_wait);
  207. BUG_ON(atomic_read(&ic->i_signaled_sends) < 0);
  208. }
  209. /*
  210. * The _oldest/_free ring operations here race cleanly with the alloc/unalloc
  211. * operations performed in the send path. As the sender allocs and potentially
  212. * unallocs the next free entry in the ring it doesn't alter which is
  213. * the next to be freed, which is what this is concerned with.
  214. */
  215. void rds_ib_send_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc)
  216. {
  217. struct rds_message *rm = NULL;
  218. struct rds_connection *conn = ic->conn;
  219. struct rds_ib_send_work *send;
  220. u32 completed;
  221. u32 oldest;
  222. u32 i = 0;
  223. int nr_sig = 0;
  224. rdsdebug("wc wr_id 0x%llx status %u (%s) byte_len %u imm_data %u\n",
  225. (unsigned long long)wc->wr_id, wc->status,
  226. ib_wc_status_msg(wc->status), wc->byte_len,
  227. be32_to_cpu(wc->ex.imm_data));
  228. rds_ib_stats_inc(s_ib_tx_cq_event);
  229. if (wc->wr_id == RDS_IB_ACK_WR_ID) {
  230. if (time_after(jiffies, ic->i_ack_queued + HZ / 2))
  231. rds_ib_stats_inc(s_ib_tx_stalled);
  232. rds_ib_ack_send_complete(ic);
  233. return;
  234. }
  235. oldest = rds_ib_ring_oldest(&ic->i_send_ring);
  236. completed = rds_ib_ring_completed(&ic->i_send_ring, wc->wr_id, oldest);
  237. for (i = 0; i < completed; i++) {
  238. send = &ic->i_sends[oldest];
  239. if (send->s_wr.send_flags & IB_SEND_SIGNALED)
  240. nr_sig++;
  241. rm = rds_ib_send_unmap_op(ic, send, wc->status);
  242. if (time_after(jiffies, send->s_queued + HZ / 2))
  243. rds_ib_stats_inc(s_ib_tx_stalled);
  244. if (send->s_op) {
  245. if (send->s_op == rm->m_final_op) {
  246. /* If anyone waited for this message to get
  247. * flushed out, wake them up now
  248. */
  249. rds_message_unmapped(rm);
  250. }
  251. rds_message_put(rm);
  252. send->s_op = NULL;
  253. }
  254. oldest = (oldest + 1) % ic->i_send_ring.w_nr;
  255. }
  256. rds_ib_ring_free(&ic->i_send_ring, completed);
  257. rds_ib_sub_signaled(ic, nr_sig);
  258. nr_sig = 0;
  259. if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags) ||
  260. test_bit(0, &conn->c_map_queued))
  261. queue_delayed_work(rds_wq, &conn->c_send_w, 0);
  262. /* We expect errors as the qp is drained during shutdown */
  263. if (wc->status != IB_WC_SUCCESS && rds_conn_up(conn)) {
  264. rds_ib_conn_error(conn, "send completion on <%pI6c,%pI6c,%d> had status %u (%s), vendor err 0x%x, disconnecting and reconnecting\n",
  265. &conn->c_laddr, &conn->c_faddr,
  266. conn->c_tos, wc->status,
  267. ib_wc_status_msg(wc->status), wc->vendor_err);
  268. }
  269. }
  270. /*
  271. * This is the main function for allocating credits when sending
  272. * messages.
  273. *
  274. * Conceptually, we have two counters:
  275. * - send credits: this tells us how many WRs we're allowed
  276. * to submit without overruning the receiver's queue. For
  277. * each SEND WR we post, we decrement this by one.
  278. *
  279. * - posted credits: this tells us how many WRs we recently
  280. * posted to the receive queue. This value is transferred
  281. * to the peer as a "credit update" in a RDS header field.
  282. * Every time we transmit credits to the peer, we subtract
  283. * the amount of transferred credits from this counter.
  284. *
  285. * It is essential that we avoid situations where both sides have
  286. * exhausted their send credits, and are unable to send new credits
  287. * to the peer. We achieve this by requiring that we send at least
  288. * one credit update to the peer before exhausting our credits.
  289. * When new credits arrive, we subtract one credit that is withheld
  290. * until we've posted new buffers and are ready to transmit these
  291. * credits (see rds_ib_send_add_credits below).
  292. *
  293. * The RDS send code is essentially single-threaded; rds_send_xmit
  294. * sets RDS_IN_XMIT to ensure exclusive access to the send ring.
  295. * However, the ACK sending code is independent and can race with
  296. * message SENDs.
  297. *
  298. * In the send path, we need to update the counters for send credits
  299. * and the counter of posted buffers atomically - when we use the
  300. * last available credit, we cannot allow another thread to race us
  301. * and grab the posted credits counter. Hence, we have to use a
  302. * spinlock to protect the credit counter, or use atomics.
  303. *
  304. * Spinlocks shared between the send and the receive path are bad,
  305. * because they create unnecessary delays. An early implementation
  306. * using a spinlock showed a 5% degradation in throughput at some
  307. * loads.
  308. *
  309. * This implementation avoids spinlocks completely, putting both
  310. * counters into a single atomic, and updating that atomic using
  311. * atomic_add (in the receive path, when receiving fresh credits),
  312. * and using atomic_cmpxchg when updating the two counters.
  313. */
  314. int rds_ib_send_grab_credits(struct rds_ib_connection *ic,
  315. u32 wanted, u32 *adv_credits, int need_posted, int max_posted)
  316. {
  317. unsigned int avail, posted, got = 0, advertise;
  318. long oldval, newval;
  319. *adv_credits = 0;
  320. if (!ic->i_flowctl)
  321. return wanted;
  322. try_again:
  323. advertise = 0;
  324. oldval = newval = atomic_read(&ic->i_credits);
  325. posted = IB_GET_POST_CREDITS(oldval);
  326. avail = IB_GET_SEND_CREDITS(oldval);
  327. rdsdebug("wanted=%u credits=%u posted=%u\n",
  328. wanted, avail, posted);
  329. /* The last credit must be used to send a credit update. */
  330. if (avail && !posted)
  331. avail--;
  332. if (avail < wanted) {
  333. struct rds_connection *conn = ic->i_cm_id->context;
  334. /* Oops, there aren't that many credits left! */
  335. set_bit(RDS_LL_SEND_FULL, &conn->c_flags);
  336. got = avail;
  337. } else {
  338. /* Sometimes you get what you want, lalala. */
  339. got = wanted;
  340. }
  341. newval -= IB_SET_SEND_CREDITS(got);
  342. /*
  343. * If need_posted is non-zero, then the caller wants
  344. * the posted regardless of whether any send credits are
  345. * available.
  346. */
  347. if (posted && (got || need_posted)) {
  348. advertise = min_t(unsigned int, posted, max_posted);
  349. newval -= IB_SET_POST_CREDITS(advertise);
  350. }
  351. /* Finally bill everything */
  352. if (atomic_cmpxchg(&ic->i_credits, oldval, newval) != oldval)
  353. goto try_again;
  354. *adv_credits = advertise;
  355. return got;
  356. }
  357. void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits)
  358. {
  359. struct rds_ib_connection *ic = conn->c_transport_data;
  360. if (credits == 0)
  361. return;
  362. rdsdebug("credits=%u current=%u%s\n",
  363. credits,
  364. IB_GET_SEND_CREDITS(atomic_read(&ic->i_credits)),
  365. test_bit(RDS_LL_SEND_FULL, &conn->c_flags) ? ", ll_send_full" : "");
  366. atomic_add(IB_SET_SEND_CREDITS(credits), &ic->i_credits);
  367. if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags))
  368. queue_delayed_work(rds_wq, &conn->c_send_w, 0);
  369. WARN_ON(IB_GET_SEND_CREDITS(credits) >= 16384);
  370. rds_ib_stats_inc(s_ib_rx_credit_updates);
  371. }
  372. void rds_ib_advertise_credits(struct rds_connection *conn, unsigned int posted)
  373. {
  374. struct rds_ib_connection *ic = conn->c_transport_data;
  375. if (posted == 0)
  376. return;
  377. atomic_add(IB_SET_POST_CREDITS(posted), &ic->i_credits);
  378. /* Decide whether to send an update to the peer now.
  379. * If we would send a credit update for every single buffer we
  380. * post, we would end up with an ACK storm (ACK arrives,
  381. * consumes buffer, we refill the ring, send ACK to remote
  382. * advertising the newly posted buffer... ad inf)
  383. *
  384. * Performance pretty much depends on how often we send
  385. * credit updates - too frequent updates mean lots of ACKs.
  386. * Too infrequent updates, and the peer will run out of
  387. * credits and has to throttle.
  388. * For the time being, 16 seems to be a good compromise.
  389. */
  390. if (IB_GET_POST_CREDITS(atomic_read(&ic->i_credits)) >= 16)
  391. set_bit(IB_ACK_REQUESTED, &ic->i_ack_flags);
  392. }
  393. static inline int rds_ib_set_wr_signal_state(struct rds_ib_connection *ic,
  394. struct rds_ib_send_work *send,
  395. bool notify)
  396. {
  397. /*
  398. * We want to delay signaling completions just enough to get
  399. * the batching benefits but not so much that we create dead time
  400. * on the wire.
  401. */
  402. if (ic->i_unsignaled_wrs-- == 0 || notify) {
  403. ic->i_unsignaled_wrs = rds_ib_sysctl_max_unsig_wrs;
  404. send->s_wr.send_flags |= IB_SEND_SIGNALED;
  405. return 1;
  406. }
  407. return 0;
  408. }
  409. /*
  410. * This can be called multiple times for a given message. The first time
  411. * we see a message we map its scatterlist into the IB device so that
  412. * we can provide that mapped address to the IB scatter gather entries
  413. * in the IB work requests. We translate the scatterlist into a series
  414. * of work requests that fragment the message. These work requests complete
  415. * in order so we pass ownership of the message to the completion handler
  416. * once we send the final fragment.
  417. *
  418. * The RDS core uses the c_send_lock to only enter this function once
  419. * per connection. This makes sure that the tx ring alloc/unalloc pairs
  420. * don't get out of sync and confuse the ring.
  421. */
  422. int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
  423. unsigned int hdr_off, unsigned int sg, unsigned int off)
  424. {
  425. struct rds_ib_connection *ic = conn->c_transport_data;
  426. struct ib_device *dev = ic->i_cm_id->device;
  427. struct rds_ib_send_work *send = NULL;
  428. struct rds_ib_send_work *first;
  429. struct rds_ib_send_work *prev;
  430. const struct ib_send_wr *failed_wr;
  431. struct scatterlist *scat;
  432. u32 pos;
  433. u32 i;
  434. u32 work_alloc;
  435. u32 credit_alloc = 0;
  436. u32 posted;
  437. u32 adv_credits = 0;
  438. int send_flags = 0;
  439. int bytes_sent = 0;
  440. int ret;
  441. int flow_controlled = 0;
  442. int nr_sig = 0;
  443. BUG_ON(off % RDS_FRAG_SIZE);
  444. BUG_ON(hdr_off != 0 && hdr_off != sizeof(struct rds_header));
  445. /* Do not send cong updates to IB loopback */
  446. if (conn->c_loopback
  447. && rm->m_inc.i_hdr.h_flags & RDS_FLAG_CONG_BITMAP) {
  448. rds_cong_map_updated(conn->c_fcong, ~(u64) 0);
  449. scat = &rm->data.op_sg[sg];
  450. ret = max_t(int, RDS_CONG_MAP_BYTES, scat->length);
  451. return sizeof(struct rds_header) + ret;
  452. }
  453. /* FIXME we may overallocate here */
  454. if (be32_to_cpu(rm->m_inc.i_hdr.h_len) == 0)
  455. i = 1;
  456. else
  457. i = DIV_ROUND_UP(be32_to_cpu(rm->m_inc.i_hdr.h_len), RDS_FRAG_SIZE);
  458. work_alloc = rds_ib_ring_alloc(&ic->i_send_ring, i, &pos);
  459. if (work_alloc == 0) {
  460. set_bit(RDS_LL_SEND_FULL, &conn->c_flags);
  461. rds_ib_stats_inc(s_ib_tx_ring_full);
  462. ret = -ENOMEM;
  463. goto out;
  464. }
  465. if (ic->i_flowctl) {
  466. credit_alloc = rds_ib_send_grab_credits(ic, work_alloc, &posted, 0, RDS_MAX_ADV_CREDIT);
  467. adv_credits += posted;
  468. if (credit_alloc < work_alloc) {
  469. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - credit_alloc);
  470. work_alloc = credit_alloc;
  471. flow_controlled = 1;
  472. }
  473. if (work_alloc == 0) {
  474. set_bit(RDS_LL_SEND_FULL, &conn->c_flags);
  475. rds_ib_stats_inc(s_ib_tx_throttle);
  476. ret = -ENOMEM;
  477. goto out;
  478. }
  479. }
  480. /* map the message the first time we see it */
  481. if (!ic->i_data_op) {
  482. if (rm->data.op_nents) {
  483. rm->data.op_count = ib_dma_map_sg(dev,
  484. rm->data.op_sg,
  485. rm->data.op_nents,
  486. DMA_TO_DEVICE);
  487. rdsdebug("ic %p mapping rm %p: %d\n", ic, rm, rm->data.op_count);
  488. if (rm->data.op_count == 0) {
  489. rds_ib_stats_inc(s_ib_tx_sg_mapping_failure);
  490. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  491. ret = -ENOMEM; /* XXX ? */
  492. goto out;
  493. }
  494. } else {
  495. rm->data.op_count = 0;
  496. }
  497. rds_message_addref(rm);
  498. rm->data.op_dmasg = 0;
  499. rm->data.op_dmaoff = 0;
  500. ic->i_data_op = &rm->data;
  501. /* Finalize the header */
  502. if (test_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags))
  503. rm->m_inc.i_hdr.h_flags |= RDS_FLAG_ACK_REQUIRED;
  504. if (test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags))
  505. rm->m_inc.i_hdr.h_flags |= RDS_FLAG_RETRANSMITTED;
  506. /* If it has a RDMA op, tell the peer we did it. This is
  507. * used by the peer to release use-once RDMA MRs. */
  508. if (rm->rdma.op_active) {
  509. struct rds_ext_header_rdma ext_hdr;
  510. ext_hdr.h_rdma_rkey = cpu_to_be32(rm->rdma.op_rkey);
  511. rds_message_add_extension(&rm->m_inc.i_hdr,
  512. RDS_EXTHDR_RDMA, &ext_hdr, sizeof(ext_hdr));
  513. }
  514. if (rm->m_rdma_cookie) {
  515. rds_message_add_rdma_dest_extension(&rm->m_inc.i_hdr,
  516. rds_rdma_cookie_key(rm->m_rdma_cookie),
  517. rds_rdma_cookie_offset(rm->m_rdma_cookie));
  518. }
  519. /* Note - rds_ib_piggyb_ack clears the ACK_REQUIRED bit, so
  520. * we should not do this unless we have a chance of at least
  521. * sticking the header into the send ring. Which is why we
  522. * should call rds_ib_ring_alloc first. */
  523. rm->m_inc.i_hdr.h_ack = cpu_to_be64(rds_ib_piggyb_ack(ic));
  524. rds_message_make_checksum(&rm->m_inc.i_hdr);
  525. /*
  526. * Update adv_credits since we reset the ACK_REQUIRED bit.
  527. */
  528. if (ic->i_flowctl) {
  529. rds_ib_send_grab_credits(ic, 0, &posted, 1, RDS_MAX_ADV_CREDIT - adv_credits);
  530. adv_credits += posted;
  531. BUG_ON(adv_credits > 255);
  532. }
  533. }
  534. /* Sometimes you want to put a fence between an RDMA
  535. * READ and the following SEND.
  536. * We could either do this all the time
  537. * or when requested by the user. Right now, we let
  538. * the application choose.
  539. */
  540. if (rm->rdma.op_active && rm->rdma.op_fence)
  541. send_flags = IB_SEND_FENCE;
  542. /* Each frag gets a header. Msgs may be 0 bytes */
  543. send = &ic->i_sends[pos];
  544. first = send;
  545. prev = NULL;
  546. scat = &ic->i_data_op->op_sg[rm->data.op_dmasg];
  547. i = 0;
  548. do {
  549. unsigned int len = 0;
  550. /* Set up the header */
  551. send->s_wr.send_flags = send_flags;
  552. send->s_wr.opcode = IB_WR_SEND;
  553. send->s_wr.num_sge = 1;
  554. send->s_wr.next = NULL;
  555. send->s_queued = jiffies;
  556. send->s_op = NULL;
  557. send->s_sge[0].addr = ic->i_send_hdrs_dma[pos];
  558. send->s_sge[0].length = sizeof(struct rds_header);
  559. send->s_sge[0].lkey = ic->i_pd->local_dma_lkey;
  560. ib_dma_sync_single_for_cpu(ic->rds_ibdev->dev,
  561. ic->i_send_hdrs_dma[pos],
  562. sizeof(struct rds_header),
  563. DMA_TO_DEVICE);
  564. memcpy(ic->i_send_hdrs[pos], &rm->m_inc.i_hdr,
  565. sizeof(struct rds_header));
  566. /* Set up the data, if present */
  567. if (i < work_alloc
  568. && scat != &rm->data.op_sg[rm->data.op_count]) {
  569. len = min(RDS_FRAG_SIZE,
  570. sg_dma_len(scat) - rm->data.op_dmaoff);
  571. send->s_wr.num_sge = 2;
  572. send->s_sge[1].addr = sg_dma_address(scat);
  573. send->s_sge[1].addr += rm->data.op_dmaoff;
  574. send->s_sge[1].length = len;
  575. send->s_sge[1].lkey = ic->i_pd->local_dma_lkey;
  576. bytes_sent += len;
  577. rm->data.op_dmaoff += len;
  578. if (rm->data.op_dmaoff == sg_dma_len(scat)) {
  579. scat++;
  580. rm->data.op_dmasg++;
  581. rm->data.op_dmaoff = 0;
  582. }
  583. }
  584. rds_ib_set_wr_signal_state(ic, send, false);
  585. /*
  586. * Always signal the last one if we're stopping due to flow control.
  587. */
  588. if (ic->i_flowctl && flow_controlled && i == (work_alloc - 1)) {
  589. rds_ib_set_wr_signal_state(ic, send, true);
  590. send->s_wr.send_flags |= IB_SEND_SOLICITED;
  591. }
  592. if (send->s_wr.send_flags & IB_SEND_SIGNALED)
  593. nr_sig++;
  594. rdsdebug("send %p wr %p num_sge %u next %p\n", send,
  595. &send->s_wr, send->s_wr.num_sge, send->s_wr.next);
  596. if (ic->i_flowctl && adv_credits) {
  597. struct rds_header *hdr = ic->i_send_hdrs[pos];
  598. /* add credit and redo the header checksum */
  599. hdr->h_credit = adv_credits;
  600. rds_message_make_checksum(hdr);
  601. adv_credits = 0;
  602. rds_ib_stats_inc(s_ib_tx_credit_updates);
  603. }
  604. ib_dma_sync_single_for_device(ic->rds_ibdev->dev,
  605. ic->i_send_hdrs_dma[pos],
  606. sizeof(struct rds_header),
  607. DMA_TO_DEVICE);
  608. if (prev)
  609. prev->s_wr.next = &send->s_wr;
  610. prev = send;
  611. pos = (pos + 1) % ic->i_send_ring.w_nr;
  612. send = &ic->i_sends[pos];
  613. i++;
  614. } while (i < work_alloc
  615. && scat != &rm->data.op_sg[rm->data.op_count]);
  616. /* Account the RDS header in the number of bytes we sent, but just once.
  617. * The caller has no concept of fragmentation. */
  618. if (hdr_off == 0)
  619. bytes_sent += sizeof(struct rds_header);
  620. /* if we finished the message then send completion owns it */
  621. if (scat == &rm->data.op_sg[rm->data.op_count]) {
  622. prev->s_op = ic->i_data_op;
  623. prev->s_wr.send_flags |= IB_SEND_SOLICITED;
  624. if (!(prev->s_wr.send_flags & IB_SEND_SIGNALED))
  625. nr_sig += rds_ib_set_wr_signal_state(ic, prev, true);
  626. ic->i_data_op = NULL;
  627. }
  628. /* Put back wrs & credits we didn't use */
  629. if (i < work_alloc) {
  630. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - i);
  631. work_alloc = i;
  632. }
  633. if (ic->i_flowctl && i < credit_alloc)
  634. rds_ib_send_add_credits(conn, credit_alloc - i);
  635. if (nr_sig)
  636. atomic_add(nr_sig, &ic->i_signaled_sends);
  637. /* XXX need to worry about failed_wr and partial sends. */
  638. failed_wr = &first->s_wr;
  639. ret = ib_post_send(ic->i_cm_id->qp, &first->s_wr, &failed_wr);
  640. rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic,
  641. first, &first->s_wr, ret, failed_wr);
  642. BUG_ON(failed_wr != &first->s_wr);
  643. if (ret) {
  644. printk(KERN_WARNING "RDS/IB: ib_post_send to %pI6c "
  645. "returned %d\n", &conn->c_faddr, ret);
  646. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  647. rds_ib_sub_signaled(ic, nr_sig);
  648. if (prev->s_op) {
  649. ic->i_data_op = prev->s_op;
  650. prev->s_op = NULL;
  651. }
  652. rds_ib_conn_error(ic->conn, "ib_post_send failed\n");
  653. goto out;
  654. }
  655. ret = bytes_sent;
  656. out:
  657. BUG_ON(adv_credits);
  658. return ret;
  659. }
  660. /*
  661. * Issue atomic operation.
  662. * A simplified version of the rdma case, we always map 1 SG, and
  663. * only 8 bytes, for the return value from the atomic operation.
  664. */
  665. int rds_ib_xmit_atomic(struct rds_connection *conn, struct rm_atomic_op *op)
  666. {
  667. struct rds_ib_connection *ic = conn->c_transport_data;
  668. struct rds_ib_send_work *send = NULL;
  669. const struct ib_send_wr *failed_wr;
  670. u32 pos;
  671. u32 work_alloc;
  672. int ret;
  673. int nr_sig = 0;
  674. work_alloc = rds_ib_ring_alloc(&ic->i_send_ring, 1, &pos);
  675. if (work_alloc != 1) {
  676. rds_ib_stats_inc(s_ib_tx_ring_full);
  677. ret = -ENOMEM;
  678. goto out;
  679. }
  680. /* address of send request in ring */
  681. send = &ic->i_sends[pos];
  682. send->s_queued = jiffies;
  683. if (op->op_type == RDS_ATOMIC_TYPE_CSWP) {
  684. send->s_atomic_wr.wr.opcode = IB_WR_MASKED_ATOMIC_CMP_AND_SWP;
  685. send->s_atomic_wr.compare_add = op->op_m_cswp.compare;
  686. send->s_atomic_wr.swap = op->op_m_cswp.swap;
  687. send->s_atomic_wr.compare_add_mask = op->op_m_cswp.compare_mask;
  688. send->s_atomic_wr.swap_mask = op->op_m_cswp.swap_mask;
  689. } else { /* FADD */
  690. send->s_atomic_wr.wr.opcode = IB_WR_MASKED_ATOMIC_FETCH_AND_ADD;
  691. send->s_atomic_wr.compare_add = op->op_m_fadd.add;
  692. send->s_atomic_wr.swap = 0;
  693. send->s_atomic_wr.compare_add_mask = op->op_m_fadd.nocarry_mask;
  694. send->s_atomic_wr.swap_mask = 0;
  695. }
  696. send->s_wr.send_flags = 0;
  697. nr_sig = rds_ib_set_wr_signal_state(ic, send, op->op_notify);
  698. send->s_atomic_wr.wr.num_sge = 1;
  699. send->s_atomic_wr.wr.next = NULL;
  700. send->s_atomic_wr.remote_addr = op->op_remote_addr;
  701. send->s_atomic_wr.rkey = op->op_rkey;
  702. send->s_op = op;
  703. rds_message_addref(container_of(send->s_op, struct rds_message, atomic));
  704. /* map 8 byte retval buffer to the device */
  705. ret = ib_dma_map_sg(ic->i_cm_id->device, op->op_sg, 1, DMA_FROM_DEVICE);
  706. rdsdebug("ic %p mapping atomic op %p. mapped %d pg\n", ic, op, ret);
  707. if (ret != 1) {
  708. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  709. rds_ib_stats_inc(s_ib_tx_sg_mapping_failure);
  710. ret = -ENOMEM; /* XXX ? */
  711. goto out;
  712. }
  713. /* Convert our struct scatterlist to struct ib_sge */
  714. send->s_sge[0].addr = sg_dma_address(op->op_sg);
  715. send->s_sge[0].length = sg_dma_len(op->op_sg);
  716. send->s_sge[0].lkey = ic->i_pd->local_dma_lkey;
  717. rdsdebug("rva %Lx rpa %Lx len %u\n", op->op_remote_addr,
  718. send->s_sge[0].addr, send->s_sge[0].length);
  719. if (nr_sig)
  720. atomic_add(nr_sig, &ic->i_signaled_sends);
  721. failed_wr = &send->s_atomic_wr.wr;
  722. ret = ib_post_send(ic->i_cm_id->qp, &send->s_atomic_wr.wr, &failed_wr);
  723. rdsdebug("ic %p send %p (wr %p) ret %d wr %p\n", ic,
  724. send, &send->s_atomic_wr, ret, failed_wr);
  725. BUG_ON(failed_wr != &send->s_atomic_wr.wr);
  726. if (ret) {
  727. printk(KERN_WARNING "RDS/IB: atomic ib_post_send to %pI6c "
  728. "returned %d\n", &conn->c_faddr, ret);
  729. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  730. rds_ib_sub_signaled(ic, nr_sig);
  731. goto out;
  732. }
  733. if (unlikely(failed_wr != &send->s_atomic_wr.wr)) {
  734. printk(KERN_WARNING "RDS/IB: atomic ib_post_send() rc=%d, but failed_wqe updated!\n", ret);
  735. BUG_ON(failed_wr != &send->s_atomic_wr.wr);
  736. }
  737. out:
  738. return ret;
  739. }
  740. int rds_ib_xmit_rdma(struct rds_connection *conn, struct rm_rdma_op *op)
  741. {
  742. struct rds_ib_connection *ic = conn->c_transport_data;
  743. struct rds_ib_send_work *send = NULL;
  744. struct rds_ib_send_work *first;
  745. struct rds_ib_send_work *prev;
  746. const struct ib_send_wr *failed_wr;
  747. struct scatterlist *scat;
  748. unsigned long len;
  749. u64 remote_addr = op->op_remote_addr;
  750. u32 max_sge = ic->rds_ibdev->max_sge;
  751. u32 pos;
  752. u32 work_alloc;
  753. u32 i;
  754. u32 j;
  755. int sent;
  756. int ret;
  757. int num_sge;
  758. int nr_sig = 0;
  759. u64 odp_addr = op->op_odp_addr;
  760. u32 odp_lkey = 0;
  761. /* map the op the first time we see it */
  762. if (!op->op_odp_mr) {
  763. if (!op->op_mapped) {
  764. op->op_count =
  765. ib_dma_map_sg(ic->i_cm_id->device, op->op_sg,
  766. op->op_nents,
  767. (op->op_write) ? DMA_TO_DEVICE :
  768. DMA_FROM_DEVICE);
  769. rdsdebug("ic %p mapping op %p: %d\n", ic, op,
  770. op->op_count);
  771. if (op->op_count == 0) {
  772. rds_ib_stats_inc(s_ib_tx_sg_mapping_failure);
  773. ret = -ENOMEM; /* XXX ? */
  774. goto out;
  775. }
  776. op->op_mapped = 1;
  777. }
  778. } else {
  779. op->op_count = op->op_nents;
  780. odp_lkey = rds_ib_get_lkey(op->op_odp_mr->r_trans_private);
  781. }
  782. /*
  783. * Instead of knowing how to return a partial rdma read/write we insist that there
  784. * be enough work requests to send the entire message.
  785. */
  786. i = DIV_ROUND_UP(op->op_count, max_sge);
  787. work_alloc = rds_ib_ring_alloc(&ic->i_send_ring, i, &pos);
  788. if (work_alloc != i) {
  789. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  790. rds_ib_stats_inc(s_ib_tx_ring_full);
  791. ret = -ENOMEM;
  792. goto out;
  793. }
  794. send = &ic->i_sends[pos];
  795. first = send;
  796. prev = NULL;
  797. scat = &op->op_sg[0];
  798. sent = 0;
  799. num_sge = op->op_count;
  800. for (i = 0; i < work_alloc && scat != &op->op_sg[op->op_count]; i++) {
  801. send->s_wr.send_flags = 0;
  802. send->s_queued = jiffies;
  803. send->s_op = NULL;
  804. if (!op->op_notify)
  805. nr_sig += rds_ib_set_wr_signal_state(ic, send,
  806. op->op_notify);
  807. send->s_wr.opcode = op->op_write ? IB_WR_RDMA_WRITE : IB_WR_RDMA_READ;
  808. send->s_rdma_wr.remote_addr = remote_addr;
  809. send->s_rdma_wr.rkey = op->op_rkey;
  810. if (num_sge > max_sge) {
  811. send->s_rdma_wr.wr.num_sge = max_sge;
  812. num_sge -= max_sge;
  813. } else {
  814. send->s_rdma_wr.wr.num_sge = num_sge;
  815. }
  816. send->s_rdma_wr.wr.next = NULL;
  817. if (prev)
  818. prev->s_rdma_wr.wr.next = &send->s_rdma_wr.wr;
  819. for (j = 0; j < send->s_rdma_wr.wr.num_sge &&
  820. scat != &op->op_sg[op->op_count]; j++) {
  821. len = sg_dma_len(scat);
  822. if (!op->op_odp_mr) {
  823. send->s_sge[j].addr = sg_dma_address(scat);
  824. send->s_sge[j].lkey = ic->i_pd->local_dma_lkey;
  825. } else {
  826. send->s_sge[j].addr = odp_addr;
  827. send->s_sge[j].lkey = odp_lkey;
  828. }
  829. send->s_sge[j].length = len;
  830. sent += len;
  831. rdsdebug("ic %p sent %d remote_addr %llu\n", ic, sent, remote_addr);
  832. remote_addr += len;
  833. odp_addr += len;
  834. scat++;
  835. }
  836. rdsdebug("send %p wr %p num_sge %u next %p\n", send,
  837. &send->s_rdma_wr.wr,
  838. send->s_rdma_wr.wr.num_sge,
  839. send->s_rdma_wr.wr.next);
  840. prev = send;
  841. if (++send == &ic->i_sends[ic->i_send_ring.w_nr])
  842. send = ic->i_sends;
  843. }
  844. /* give a reference to the last op */
  845. if (scat == &op->op_sg[op->op_count]) {
  846. prev->s_op = op;
  847. rds_message_addref(container_of(op, struct rds_message, rdma));
  848. }
  849. if (i < work_alloc) {
  850. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - i);
  851. work_alloc = i;
  852. }
  853. if (nr_sig)
  854. atomic_add(nr_sig, &ic->i_signaled_sends);
  855. failed_wr = &first->s_rdma_wr.wr;
  856. ret = ib_post_send(ic->i_cm_id->qp, &first->s_rdma_wr.wr, &failed_wr);
  857. rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic,
  858. first, &first->s_rdma_wr.wr, ret, failed_wr);
  859. BUG_ON(failed_wr != &first->s_rdma_wr.wr);
  860. if (ret) {
  861. printk(KERN_WARNING "RDS/IB: rdma ib_post_send to %pI6c "
  862. "returned %d\n", &conn->c_faddr, ret);
  863. rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
  864. rds_ib_sub_signaled(ic, nr_sig);
  865. goto out;
  866. }
  867. if (unlikely(failed_wr != &first->s_rdma_wr.wr)) {
  868. printk(KERN_WARNING "RDS/IB: ib_post_send() rc=%d, but failed_wqe updated!\n", ret);
  869. BUG_ON(failed_wr != &first->s_rdma_wr.wr);
  870. }
  871. out:
  872. return ret;
  873. }
  874. void rds_ib_xmit_path_complete(struct rds_conn_path *cp)
  875. {
  876. struct rds_connection *conn = cp->cp_conn;
  877. struct rds_ib_connection *ic = conn->c_transport_data;
  878. /* We may have a pending ACK or window update we were unable
  879. * to send previously (due to flow control). Try again. */
  880. rds_ib_attempt_ack(ic);
  881. }