svc_rdma_transport.c 31 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088
  1. /*
  2. * Copyright (c) 2005-2007 Network Appliance, Inc. 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 BSD-type
  8. * license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or without
  11. * modification, are permitted provided that the following conditions
  12. * are met:
  13. *
  14. * Redistributions of source code must retain the above copyright
  15. * notice, this list of conditions and the following disclaimer.
  16. *
  17. * Redistributions in binary form must reproduce the above
  18. * copyright notice, this list of conditions and the following
  19. * disclaimer in the documentation and/or other materials provided
  20. * with the distribution.
  21. *
  22. * Neither the name of the Network Appliance, Inc. nor the names of
  23. * its contributors may be used to endorse or promote products
  24. * derived from this software without specific prior written
  25. * permission.
  26. *
  27. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  28. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  29. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  30. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  31. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  32. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  33. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  34. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  35. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  36. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  37. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  38. *
  39. * Author: Tom Tucker <tom@opengridcomputing.com>
  40. */
  41. #include <linux/sunrpc/svc_xprt.h>
  42. #include <linux/sunrpc/debug.h>
  43. #include <linux/sunrpc/rpc_rdma.h>
  44. #include <linux/spinlock.h>
  45. #include <rdma/ib_verbs.h>
  46. #include <rdma/rdma_cm.h>
  47. #include <linux/sunrpc/svc_rdma.h>
  48. #define RPCDBG_FACILITY RPCDBG_SVCXPRT
  49. static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
  50. struct sockaddr *sa, int salen,
  51. int flags);
  52. static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt);
  53. static void svc_rdma_release_rqst(struct svc_rqst *);
  54. static void dto_tasklet_func(unsigned long data);
  55. static void svc_rdma_detach(struct svc_xprt *xprt);
  56. static void svc_rdma_free(struct svc_xprt *xprt);
  57. static int svc_rdma_has_wspace(struct svc_xprt *xprt);
  58. static void rq_cq_reap(struct svcxprt_rdma *xprt);
  59. static void sq_cq_reap(struct svcxprt_rdma *xprt);
  60. DECLARE_TASKLET(dto_tasklet, dto_tasklet_func, 0UL);
  61. static DEFINE_SPINLOCK(dto_lock);
  62. static LIST_HEAD(dto_xprt_q);
  63. static struct svc_xprt_ops svc_rdma_ops = {
  64. .xpo_create = svc_rdma_create,
  65. .xpo_recvfrom = svc_rdma_recvfrom,
  66. .xpo_sendto = svc_rdma_sendto,
  67. .xpo_release_rqst = svc_rdma_release_rqst,
  68. .xpo_detach = svc_rdma_detach,
  69. .xpo_free = svc_rdma_free,
  70. .xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
  71. .xpo_has_wspace = svc_rdma_has_wspace,
  72. .xpo_accept = svc_rdma_accept,
  73. };
  74. struct svc_xprt_class svc_rdma_class = {
  75. .xcl_name = "rdma",
  76. .xcl_owner = THIS_MODULE,
  77. .xcl_ops = &svc_rdma_ops,
  78. .xcl_max_payload = RPCSVC_MAXPAYLOAD_TCP,
  79. };
  80. static int rdma_bump_context_cache(struct svcxprt_rdma *xprt)
  81. {
  82. int target;
  83. int at_least_one = 0;
  84. struct svc_rdma_op_ctxt *ctxt;
  85. target = min(xprt->sc_ctxt_cnt + xprt->sc_ctxt_bump,
  86. xprt->sc_ctxt_max);
  87. spin_lock_bh(&xprt->sc_ctxt_lock);
  88. while (xprt->sc_ctxt_cnt < target) {
  89. xprt->sc_ctxt_cnt++;
  90. spin_unlock_bh(&xprt->sc_ctxt_lock);
  91. ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL);
  92. spin_lock_bh(&xprt->sc_ctxt_lock);
  93. if (ctxt) {
  94. at_least_one = 1;
  95. ctxt->next = xprt->sc_ctxt_head;
  96. xprt->sc_ctxt_head = ctxt;
  97. } else {
  98. /* kmalloc failed...give up for now */
  99. xprt->sc_ctxt_cnt--;
  100. break;
  101. }
  102. }
  103. spin_unlock_bh(&xprt->sc_ctxt_lock);
  104. dprintk("svcrdma: sc_ctxt_max=%d, sc_ctxt_cnt=%d\n",
  105. xprt->sc_ctxt_max, xprt->sc_ctxt_cnt);
  106. return at_least_one;
  107. }
  108. struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
  109. {
  110. struct svc_rdma_op_ctxt *ctxt;
  111. while (1) {
  112. spin_lock_bh(&xprt->sc_ctxt_lock);
  113. if (unlikely(xprt->sc_ctxt_head == NULL)) {
  114. /* Try to bump my cache. */
  115. spin_unlock_bh(&xprt->sc_ctxt_lock);
  116. if (rdma_bump_context_cache(xprt))
  117. continue;
  118. printk(KERN_INFO "svcrdma: sleeping waiting for "
  119. "context memory on xprt=%p\n",
  120. xprt);
  121. schedule_timeout_uninterruptible(msecs_to_jiffies(500));
  122. continue;
  123. }
  124. ctxt = xprt->sc_ctxt_head;
  125. xprt->sc_ctxt_head = ctxt->next;
  126. spin_unlock_bh(&xprt->sc_ctxt_lock);
  127. ctxt->xprt = xprt;
  128. INIT_LIST_HEAD(&ctxt->dto_q);
  129. ctxt->count = 0;
  130. break;
  131. }
  132. return ctxt;
  133. }
  134. void svc_rdma_put_context(struct svc_rdma_op_ctxt *ctxt, int free_pages)
  135. {
  136. struct svcxprt_rdma *xprt;
  137. int i;
  138. BUG_ON(!ctxt);
  139. xprt = ctxt->xprt;
  140. if (free_pages)
  141. for (i = 0; i < ctxt->count; i++)
  142. put_page(ctxt->pages[i]);
  143. for (i = 0; i < ctxt->count; i++)
  144. dma_unmap_single(xprt->sc_cm_id->device->dma_device,
  145. ctxt->sge[i].addr,
  146. ctxt->sge[i].length,
  147. ctxt->direction);
  148. spin_lock_bh(&xprt->sc_ctxt_lock);
  149. ctxt->next = xprt->sc_ctxt_head;
  150. xprt->sc_ctxt_head = ctxt;
  151. spin_unlock_bh(&xprt->sc_ctxt_lock);
  152. }
  153. /* ib_cq event handler */
  154. static void cq_event_handler(struct ib_event *event, void *context)
  155. {
  156. struct svc_xprt *xprt = context;
  157. dprintk("svcrdma: received CQ event id=%d, context=%p\n",
  158. event->event, context);
  159. set_bit(XPT_CLOSE, &xprt->xpt_flags);
  160. }
  161. /* QP event handler */
  162. static void qp_event_handler(struct ib_event *event, void *context)
  163. {
  164. struct svc_xprt *xprt = context;
  165. switch (event->event) {
  166. /* These are considered benign events */
  167. case IB_EVENT_PATH_MIG:
  168. case IB_EVENT_COMM_EST:
  169. case IB_EVENT_SQ_DRAINED:
  170. case IB_EVENT_QP_LAST_WQE_REACHED:
  171. dprintk("svcrdma: QP event %d received for QP=%p\n",
  172. event->event, event->element.qp);
  173. break;
  174. /* These are considered fatal events */
  175. case IB_EVENT_PATH_MIG_ERR:
  176. case IB_EVENT_QP_FATAL:
  177. case IB_EVENT_QP_REQ_ERR:
  178. case IB_EVENT_QP_ACCESS_ERR:
  179. case IB_EVENT_DEVICE_FATAL:
  180. default:
  181. dprintk("svcrdma: QP ERROR event %d received for QP=%p, "
  182. "closing transport\n",
  183. event->event, event->element.qp);
  184. set_bit(XPT_CLOSE, &xprt->xpt_flags);
  185. break;
  186. }
  187. }
  188. /*
  189. * Data Transfer Operation Tasklet
  190. *
  191. * Walks a list of transports with I/O pending, removing entries as
  192. * they are added to the server's I/O pending list. Two bits indicate
  193. * if SQ, RQ, or both have I/O pending. The dto_lock is an irqsave
  194. * spinlock that serializes access to the transport list with the RQ
  195. * and SQ interrupt handlers.
  196. */
  197. static void dto_tasklet_func(unsigned long data)
  198. {
  199. struct svcxprt_rdma *xprt;
  200. unsigned long flags;
  201. spin_lock_irqsave(&dto_lock, flags);
  202. while (!list_empty(&dto_xprt_q)) {
  203. xprt = list_entry(dto_xprt_q.next,
  204. struct svcxprt_rdma, sc_dto_q);
  205. list_del_init(&xprt->sc_dto_q);
  206. spin_unlock_irqrestore(&dto_lock, flags);
  207. rq_cq_reap(xprt);
  208. sq_cq_reap(xprt);
  209. svc_xprt_put(&xprt->sc_xprt);
  210. spin_lock_irqsave(&dto_lock, flags);
  211. }
  212. spin_unlock_irqrestore(&dto_lock, flags);
  213. }
  214. /*
  215. * Receive Queue Completion Handler
  216. *
  217. * Since an RQ completion handler is called on interrupt context, we
  218. * need to defer the handling of the I/O to a tasklet
  219. */
  220. static void rq_comp_handler(struct ib_cq *cq, void *cq_context)
  221. {
  222. struct svcxprt_rdma *xprt = cq_context;
  223. unsigned long flags;
  224. /*
  225. * Set the bit regardless of whether or not it's on the list
  226. * because it may be on the list already due to an SQ
  227. * completion.
  228. */
  229. set_bit(RDMAXPRT_RQ_PENDING, &xprt->sc_flags);
  230. /*
  231. * If this transport is not already on the DTO transport queue,
  232. * add it
  233. */
  234. spin_lock_irqsave(&dto_lock, flags);
  235. if (list_empty(&xprt->sc_dto_q)) {
  236. svc_xprt_get(&xprt->sc_xprt);
  237. list_add_tail(&xprt->sc_dto_q, &dto_xprt_q);
  238. }
  239. spin_unlock_irqrestore(&dto_lock, flags);
  240. /* Tasklet does all the work to avoid irqsave locks. */
  241. tasklet_schedule(&dto_tasklet);
  242. }
  243. /*
  244. * rq_cq_reap - Process the RQ CQ.
  245. *
  246. * Take all completing WC off the CQE and enqueue the associated DTO
  247. * context on the dto_q for the transport.
  248. */
  249. static void rq_cq_reap(struct svcxprt_rdma *xprt)
  250. {
  251. int ret;
  252. struct ib_wc wc;
  253. struct svc_rdma_op_ctxt *ctxt = NULL;
  254. if (!test_and_clear_bit(RDMAXPRT_RQ_PENDING, &xprt->sc_flags))
  255. return;
  256. ib_req_notify_cq(xprt->sc_rq_cq, IB_CQ_NEXT_COMP);
  257. atomic_inc(&rdma_stat_rq_poll);
  258. spin_lock_bh(&xprt->sc_rq_dto_lock);
  259. while ((ret = ib_poll_cq(xprt->sc_rq_cq, 1, &wc)) > 0) {
  260. ctxt = (struct svc_rdma_op_ctxt *)(unsigned long)wc.wr_id;
  261. ctxt->wc_status = wc.status;
  262. ctxt->byte_len = wc.byte_len;
  263. if (wc.status != IB_WC_SUCCESS) {
  264. /* Close the transport */
  265. set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
  266. svc_rdma_put_context(ctxt, 1);
  267. continue;
  268. }
  269. list_add_tail(&ctxt->dto_q, &xprt->sc_rq_dto_q);
  270. }
  271. spin_unlock_bh(&xprt->sc_rq_dto_lock);
  272. if (ctxt)
  273. atomic_inc(&rdma_stat_rq_prod);
  274. set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
  275. /*
  276. * If data arrived before established event,
  277. * don't enqueue. This defers RPC I/O until the
  278. * RDMA connection is complete.
  279. */
  280. if (!test_bit(RDMAXPRT_CONN_PENDING, &xprt->sc_flags))
  281. svc_xprt_enqueue(&xprt->sc_xprt);
  282. }
  283. /*
  284. * Send Queue Completion Handler - potentially called on interrupt context.
  285. */
  286. static void sq_cq_reap(struct svcxprt_rdma *xprt)
  287. {
  288. struct svc_rdma_op_ctxt *ctxt = NULL;
  289. struct ib_wc wc;
  290. struct ib_cq *cq = xprt->sc_sq_cq;
  291. int ret;
  292. if (!test_and_clear_bit(RDMAXPRT_SQ_PENDING, &xprt->sc_flags))
  293. return;
  294. ib_req_notify_cq(xprt->sc_sq_cq, IB_CQ_NEXT_COMP);
  295. atomic_inc(&rdma_stat_sq_poll);
  296. while ((ret = ib_poll_cq(cq, 1, &wc)) > 0) {
  297. ctxt = (struct svc_rdma_op_ctxt *)(unsigned long)wc.wr_id;
  298. xprt = ctxt->xprt;
  299. if (wc.status != IB_WC_SUCCESS)
  300. /* Close the transport */
  301. set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
  302. /* Decrement used SQ WR count */
  303. atomic_dec(&xprt->sc_sq_count);
  304. wake_up(&xprt->sc_send_wait);
  305. switch (ctxt->wr_op) {
  306. case IB_WR_SEND:
  307. case IB_WR_RDMA_WRITE:
  308. svc_rdma_put_context(ctxt, 1);
  309. break;
  310. case IB_WR_RDMA_READ:
  311. if (test_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags)) {
  312. set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
  313. set_bit(RDMACTXT_F_READ_DONE, &ctxt->flags);
  314. spin_lock_bh(&xprt->sc_read_complete_lock);
  315. list_add_tail(&ctxt->dto_q,
  316. &xprt->sc_read_complete_q);
  317. spin_unlock_bh(&xprt->sc_read_complete_lock);
  318. svc_xprt_enqueue(&xprt->sc_xprt);
  319. }
  320. break;
  321. default:
  322. printk(KERN_ERR "svcrdma: unexpected completion type, "
  323. "opcode=%d, status=%d\n",
  324. wc.opcode, wc.status);
  325. break;
  326. }
  327. }
  328. if (ctxt)
  329. atomic_inc(&rdma_stat_sq_prod);
  330. }
  331. static void sq_comp_handler(struct ib_cq *cq, void *cq_context)
  332. {
  333. struct svcxprt_rdma *xprt = cq_context;
  334. unsigned long flags;
  335. /*
  336. * Set the bit regardless of whether or not it's on the list
  337. * because it may be on the list already due to an RQ
  338. * completion.
  339. */
  340. set_bit(RDMAXPRT_SQ_PENDING, &xprt->sc_flags);
  341. /*
  342. * If this transport is not already on the DTO transport queue,
  343. * add it
  344. */
  345. spin_lock_irqsave(&dto_lock, flags);
  346. if (list_empty(&xprt->sc_dto_q)) {
  347. svc_xprt_get(&xprt->sc_xprt);
  348. list_add_tail(&xprt->sc_dto_q, &dto_xprt_q);
  349. }
  350. spin_unlock_irqrestore(&dto_lock, flags);
  351. /* Tasklet does all the work to avoid irqsave locks. */
  352. tasklet_schedule(&dto_tasklet);
  353. }
  354. static void create_context_cache(struct svcxprt_rdma *xprt,
  355. int ctxt_count, int ctxt_bump, int ctxt_max)
  356. {
  357. struct svc_rdma_op_ctxt *ctxt;
  358. int i;
  359. xprt->sc_ctxt_max = ctxt_max;
  360. xprt->sc_ctxt_bump = ctxt_bump;
  361. xprt->sc_ctxt_cnt = 0;
  362. xprt->sc_ctxt_head = NULL;
  363. for (i = 0; i < ctxt_count; i++) {
  364. ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL);
  365. if (ctxt) {
  366. ctxt->next = xprt->sc_ctxt_head;
  367. xprt->sc_ctxt_head = ctxt;
  368. xprt->sc_ctxt_cnt++;
  369. }
  370. }
  371. }
  372. static void destroy_context_cache(struct svc_rdma_op_ctxt *ctxt)
  373. {
  374. struct svc_rdma_op_ctxt *next;
  375. if (!ctxt)
  376. return;
  377. do {
  378. next = ctxt->next;
  379. kfree(ctxt);
  380. ctxt = next;
  381. } while (next);
  382. }
  383. static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
  384. int listener)
  385. {
  386. struct svcxprt_rdma *cma_xprt = kzalloc(sizeof *cma_xprt, GFP_KERNEL);
  387. if (!cma_xprt)
  388. return NULL;
  389. svc_xprt_init(&svc_rdma_class, &cma_xprt->sc_xprt, serv);
  390. INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
  391. INIT_LIST_HEAD(&cma_xprt->sc_dto_q);
  392. INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
  393. INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
  394. init_waitqueue_head(&cma_xprt->sc_send_wait);
  395. spin_lock_init(&cma_xprt->sc_lock);
  396. spin_lock_init(&cma_xprt->sc_read_complete_lock);
  397. spin_lock_init(&cma_xprt->sc_ctxt_lock);
  398. spin_lock_init(&cma_xprt->sc_rq_dto_lock);
  399. cma_xprt->sc_ord = svcrdma_ord;
  400. cma_xprt->sc_max_req_size = svcrdma_max_req_size;
  401. cma_xprt->sc_max_requests = svcrdma_max_requests;
  402. cma_xprt->sc_sq_depth = svcrdma_max_requests * RPCRDMA_SQ_DEPTH_MULT;
  403. atomic_set(&cma_xprt->sc_sq_count, 0);
  404. if (!listener) {
  405. int reqs = cma_xprt->sc_max_requests;
  406. create_context_cache(cma_xprt,
  407. reqs << 1, /* starting size */
  408. reqs, /* bump amount */
  409. reqs +
  410. cma_xprt->sc_sq_depth +
  411. RPCRDMA_MAX_THREADS + 1); /* max */
  412. if (!cma_xprt->sc_ctxt_head) {
  413. kfree(cma_xprt);
  414. return NULL;
  415. }
  416. clear_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
  417. } else
  418. set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
  419. return cma_xprt;
  420. }
  421. struct page *svc_rdma_get_page(void)
  422. {
  423. struct page *page;
  424. while ((page = alloc_page(GFP_KERNEL)) == NULL) {
  425. /* If we can't get memory, wait a bit and try again */
  426. printk(KERN_INFO "svcrdma: out of memory...retrying in 1000 "
  427. "jiffies.\n");
  428. schedule_timeout_uninterruptible(msecs_to_jiffies(1000));
  429. }
  430. return page;
  431. }
  432. int svc_rdma_post_recv(struct svcxprt_rdma *xprt)
  433. {
  434. struct ib_recv_wr recv_wr, *bad_recv_wr;
  435. struct svc_rdma_op_ctxt *ctxt;
  436. struct page *page;
  437. unsigned long pa;
  438. int sge_no;
  439. int buflen;
  440. int ret;
  441. ctxt = svc_rdma_get_context(xprt);
  442. buflen = 0;
  443. ctxt->direction = DMA_FROM_DEVICE;
  444. for (sge_no = 0; buflen < xprt->sc_max_req_size; sge_no++) {
  445. BUG_ON(sge_no >= xprt->sc_max_sge);
  446. page = svc_rdma_get_page();
  447. ctxt->pages[sge_no] = page;
  448. pa = ib_dma_map_page(xprt->sc_cm_id->device,
  449. page, 0, PAGE_SIZE,
  450. DMA_FROM_DEVICE);
  451. ctxt->sge[sge_no].addr = pa;
  452. ctxt->sge[sge_no].length = PAGE_SIZE;
  453. ctxt->sge[sge_no].lkey = xprt->sc_phys_mr->lkey;
  454. buflen += PAGE_SIZE;
  455. }
  456. ctxt->count = sge_no;
  457. recv_wr.next = NULL;
  458. recv_wr.sg_list = &ctxt->sge[0];
  459. recv_wr.num_sge = ctxt->count;
  460. recv_wr.wr_id = (u64)(unsigned long)ctxt;
  461. ret = ib_post_recv(xprt->sc_qp, &recv_wr, &bad_recv_wr);
  462. return ret;
  463. }
  464. /*
  465. * This function handles the CONNECT_REQUEST event on a listening
  466. * endpoint. It is passed the cma_id for the _new_ connection. The context in
  467. * this cma_id is inherited from the listening cma_id and is the svc_xprt
  468. * structure for the listening endpoint.
  469. *
  470. * This function creates a new xprt for the new connection and enqueues it on
  471. * the accept queue for the listent xprt. When the listen thread is kicked, it
  472. * will call the recvfrom method on the listen xprt which will accept the new
  473. * connection.
  474. */
  475. static void handle_connect_req(struct rdma_cm_id *new_cma_id)
  476. {
  477. struct svcxprt_rdma *listen_xprt = new_cma_id->context;
  478. struct svcxprt_rdma *newxprt;
  479. /* Create a new transport */
  480. newxprt = rdma_create_xprt(listen_xprt->sc_xprt.xpt_server, 0);
  481. if (!newxprt) {
  482. dprintk("svcrdma: failed to create new transport\n");
  483. return;
  484. }
  485. newxprt->sc_cm_id = new_cma_id;
  486. new_cma_id->context = newxprt;
  487. dprintk("svcrdma: Creating newxprt=%p, cm_id=%p, listenxprt=%p\n",
  488. newxprt, newxprt->sc_cm_id, listen_xprt);
  489. /*
  490. * Enqueue the new transport on the accept queue of the listening
  491. * transport
  492. */
  493. spin_lock_bh(&listen_xprt->sc_lock);
  494. list_add_tail(&newxprt->sc_accept_q, &listen_xprt->sc_accept_q);
  495. spin_unlock_bh(&listen_xprt->sc_lock);
  496. /*
  497. * Can't use svc_xprt_received here because we are not on a
  498. * rqstp thread
  499. */
  500. set_bit(XPT_CONN, &listen_xprt->sc_xprt.xpt_flags);
  501. svc_xprt_enqueue(&listen_xprt->sc_xprt);
  502. }
  503. /*
  504. * Handles events generated on the listening endpoint. These events will be
  505. * either be incoming connect requests or adapter removal events.
  506. */
  507. static int rdma_listen_handler(struct rdma_cm_id *cma_id,
  508. struct rdma_cm_event *event)
  509. {
  510. struct svcxprt_rdma *xprt = cma_id->context;
  511. int ret = 0;
  512. switch (event->event) {
  513. case RDMA_CM_EVENT_CONNECT_REQUEST:
  514. dprintk("svcrdma: Connect request on cma_id=%p, xprt = %p, "
  515. "event=%d\n", cma_id, cma_id->context, event->event);
  516. handle_connect_req(cma_id);
  517. break;
  518. case RDMA_CM_EVENT_ESTABLISHED:
  519. /* Accept complete */
  520. dprintk("svcrdma: Connection completed on LISTEN xprt=%p, "
  521. "cm_id=%p\n", xprt, cma_id);
  522. break;
  523. case RDMA_CM_EVENT_DEVICE_REMOVAL:
  524. dprintk("svcrdma: Device removal xprt=%p, cm_id=%p\n",
  525. xprt, cma_id);
  526. if (xprt)
  527. set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
  528. break;
  529. default:
  530. dprintk("svcrdma: Unexpected event on listening endpoint %p, "
  531. "event=%d\n", cma_id, event->event);
  532. break;
  533. }
  534. return ret;
  535. }
  536. static int rdma_cma_handler(struct rdma_cm_id *cma_id,
  537. struct rdma_cm_event *event)
  538. {
  539. struct svc_xprt *xprt = cma_id->context;
  540. struct svcxprt_rdma *rdma =
  541. container_of(xprt, struct svcxprt_rdma, sc_xprt);
  542. switch (event->event) {
  543. case RDMA_CM_EVENT_ESTABLISHED:
  544. /* Accept complete */
  545. svc_xprt_get(xprt);
  546. dprintk("svcrdma: Connection completed on DTO xprt=%p, "
  547. "cm_id=%p\n", xprt, cma_id);
  548. clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
  549. svc_xprt_enqueue(xprt);
  550. break;
  551. case RDMA_CM_EVENT_DISCONNECTED:
  552. dprintk("svcrdma: Disconnect on DTO xprt=%p, cm_id=%p\n",
  553. xprt, cma_id);
  554. if (xprt) {
  555. set_bit(XPT_CLOSE, &xprt->xpt_flags);
  556. svc_xprt_enqueue(xprt);
  557. svc_xprt_put(xprt);
  558. }
  559. break;
  560. case RDMA_CM_EVENT_DEVICE_REMOVAL:
  561. dprintk("svcrdma: Device removal cma_id=%p, xprt = %p, "
  562. "event=%d\n", cma_id, xprt, event->event);
  563. if (xprt) {
  564. set_bit(XPT_CLOSE, &xprt->xpt_flags);
  565. svc_xprt_enqueue(xprt);
  566. }
  567. break;
  568. default:
  569. dprintk("svcrdma: Unexpected event on DTO endpoint %p, "
  570. "event=%d\n", cma_id, event->event);
  571. break;
  572. }
  573. return 0;
  574. }
  575. /*
  576. * Create a listening RDMA service endpoint.
  577. */
  578. static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
  579. struct sockaddr *sa, int salen,
  580. int flags)
  581. {
  582. struct rdma_cm_id *listen_id;
  583. struct svcxprt_rdma *cma_xprt;
  584. struct svc_xprt *xprt;
  585. int ret;
  586. dprintk("svcrdma: Creating RDMA socket\n");
  587. cma_xprt = rdma_create_xprt(serv, 1);
  588. if (!cma_xprt)
  589. return ERR_PTR(ENOMEM);
  590. xprt = &cma_xprt->sc_xprt;
  591. listen_id = rdma_create_id(rdma_listen_handler, cma_xprt, RDMA_PS_TCP);
  592. if (IS_ERR(listen_id)) {
  593. svc_xprt_put(&cma_xprt->sc_xprt);
  594. dprintk("svcrdma: rdma_create_id failed = %ld\n",
  595. PTR_ERR(listen_id));
  596. return (void *)listen_id;
  597. }
  598. ret = rdma_bind_addr(listen_id, sa);
  599. if (ret) {
  600. rdma_destroy_id(listen_id);
  601. svc_xprt_put(&cma_xprt->sc_xprt);
  602. dprintk("svcrdma: rdma_bind_addr failed = %d\n", ret);
  603. return ERR_PTR(ret);
  604. }
  605. cma_xprt->sc_cm_id = listen_id;
  606. ret = rdma_listen(listen_id, RPCRDMA_LISTEN_BACKLOG);
  607. if (ret) {
  608. rdma_destroy_id(listen_id);
  609. svc_xprt_put(&cma_xprt->sc_xprt);
  610. dprintk("svcrdma: rdma_listen failed = %d\n", ret);
  611. return ERR_PTR(ret);
  612. }
  613. /*
  614. * We need to use the address from the cm_id in case the
  615. * caller specified 0 for the port number.
  616. */
  617. sa = (struct sockaddr *)&cma_xprt->sc_cm_id->route.addr.src_addr;
  618. svc_xprt_set_local(&cma_xprt->sc_xprt, sa, salen);
  619. return &cma_xprt->sc_xprt;
  620. }
  621. /*
  622. * This is the xpo_recvfrom function for listening endpoints. Its
  623. * purpose is to accept incoming connections. The CMA callback handler
  624. * has already created a new transport and attached it to the new CMA
  625. * ID.
  626. *
  627. * There is a queue of pending connections hung on the listening
  628. * transport. This queue contains the new svc_xprt structure. This
  629. * function takes svc_xprt structures off the accept_q and completes
  630. * the connection.
  631. */
  632. static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
  633. {
  634. struct svcxprt_rdma *listen_rdma;
  635. struct svcxprt_rdma *newxprt = NULL;
  636. struct rdma_conn_param conn_param;
  637. struct ib_qp_init_attr qp_attr;
  638. struct ib_device_attr devattr;
  639. struct sockaddr *sa;
  640. int ret;
  641. int i;
  642. listen_rdma = container_of(xprt, struct svcxprt_rdma, sc_xprt);
  643. clear_bit(XPT_CONN, &xprt->xpt_flags);
  644. /* Get the next entry off the accept list */
  645. spin_lock_bh(&listen_rdma->sc_lock);
  646. if (!list_empty(&listen_rdma->sc_accept_q)) {
  647. newxprt = list_entry(listen_rdma->sc_accept_q.next,
  648. struct svcxprt_rdma, sc_accept_q);
  649. list_del_init(&newxprt->sc_accept_q);
  650. }
  651. if (!list_empty(&listen_rdma->sc_accept_q))
  652. set_bit(XPT_CONN, &listen_rdma->sc_xprt.xpt_flags);
  653. spin_unlock_bh(&listen_rdma->sc_lock);
  654. if (!newxprt)
  655. return NULL;
  656. dprintk("svcrdma: newxprt from accept queue = %p, cm_id=%p\n",
  657. newxprt, newxprt->sc_cm_id);
  658. ret = ib_query_device(newxprt->sc_cm_id->device, &devattr);
  659. if (ret) {
  660. dprintk("svcrdma: could not query device attributes on "
  661. "device %p, rc=%d\n", newxprt->sc_cm_id->device, ret);
  662. goto errout;
  663. }
  664. /* Qualify the transport resource defaults with the
  665. * capabilities of this particular device */
  666. newxprt->sc_max_sge = min((size_t)devattr.max_sge,
  667. (size_t)RPCSVC_MAXPAGES);
  668. newxprt->sc_max_requests = min((size_t)devattr.max_qp_wr,
  669. (size_t)svcrdma_max_requests);
  670. newxprt->sc_sq_depth = RPCRDMA_SQ_DEPTH_MULT * newxprt->sc_max_requests;
  671. newxprt->sc_ord = min((size_t)devattr.max_qp_rd_atom,
  672. (size_t)svcrdma_ord);
  673. newxprt->sc_pd = ib_alloc_pd(newxprt->sc_cm_id->device);
  674. if (IS_ERR(newxprt->sc_pd)) {
  675. dprintk("svcrdma: error creating PD for connect request\n");
  676. goto errout;
  677. }
  678. newxprt->sc_sq_cq = ib_create_cq(newxprt->sc_cm_id->device,
  679. sq_comp_handler,
  680. cq_event_handler,
  681. newxprt,
  682. newxprt->sc_sq_depth,
  683. 0);
  684. if (IS_ERR(newxprt->sc_sq_cq)) {
  685. dprintk("svcrdma: error creating SQ CQ for connect request\n");
  686. goto errout;
  687. }
  688. newxprt->sc_rq_cq = ib_create_cq(newxprt->sc_cm_id->device,
  689. rq_comp_handler,
  690. cq_event_handler,
  691. newxprt,
  692. newxprt->sc_max_requests,
  693. 0);
  694. if (IS_ERR(newxprt->sc_rq_cq)) {
  695. dprintk("svcrdma: error creating RQ CQ for connect request\n");
  696. goto errout;
  697. }
  698. memset(&qp_attr, 0, sizeof qp_attr);
  699. qp_attr.event_handler = qp_event_handler;
  700. qp_attr.qp_context = &newxprt->sc_xprt;
  701. qp_attr.cap.max_send_wr = newxprt->sc_sq_depth;
  702. qp_attr.cap.max_recv_wr = newxprt->sc_max_requests;
  703. qp_attr.cap.max_send_sge = newxprt->sc_max_sge;
  704. qp_attr.cap.max_recv_sge = newxprt->sc_max_sge;
  705. qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
  706. qp_attr.qp_type = IB_QPT_RC;
  707. qp_attr.send_cq = newxprt->sc_sq_cq;
  708. qp_attr.recv_cq = newxprt->sc_rq_cq;
  709. dprintk("svcrdma: newxprt->sc_cm_id=%p, newxprt->sc_pd=%p\n"
  710. " cm_id->device=%p, sc_pd->device=%p\n"
  711. " cap.max_send_wr = %d\n"
  712. " cap.max_recv_wr = %d\n"
  713. " cap.max_send_sge = %d\n"
  714. " cap.max_recv_sge = %d\n",
  715. newxprt->sc_cm_id, newxprt->sc_pd,
  716. newxprt->sc_cm_id->device, newxprt->sc_pd->device,
  717. qp_attr.cap.max_send_wr,
  718. qp_attr.cap.max_recv_wr,
  719. qp_attr.cap.max_send_sge,
  720. qp_attr.cap.max_recv_sge);
  721. ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd, &qp_attr);
  722. if (ret) {
  723. /*
  724. * XXX: This is a hack. We need a xx_request_qp interface
  725. * that will adjust the qp_attr's with a best-effort
  726. * number
  727. */
  728. qp_attr.cap.max_send_sge -= 2;
  729. qp_attr.cap.max_recv_sge -= 2;
  730. ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd,
  731. &qp_attr);
  732. if (ret) {
  733. dprintk("svcrdma: failed to create QP, ret=%d\n", ret);
  734. goto errout;
  735. }
  736. newxprt->sc_max_sge = qp_attr.cap.max_send_sge;
  737. newxprt->sc_max_sge = qp_attr.cap.max_recv_sge;
  738. newxprt->sc_sq_depth = qp_attr.cap.max_send_wr;
  739. newxprt->sc_max_requests = qp_attr.cap.max_recv_wr;
  740. }
  741. svc_xprt_get(&newxprt->sc_xprt);
  742. newxprt->sc_qp = newxprt->sc_cm_id->qp;
  743. /* Register all of physical memory */
  744. newxprt->sc_phys_mr = ib_get_dma_mr(newxprt->sc_pd,
  745. IB_ACCESS_LOCAL_WRITE |
  746. IB_ACCESS_REMOTE_WRITE);
  747. if (IS_ERR(newxprt->sc_phys_mr)) {
  748. dprintk("svcrdma: Failed to create DMA MR ret=%d\n", ret);
  749. goto errout;
  750. }
  751. /* Post receive buffers */
  752. for (i = 0; i < newxprt->sc_max_requests; i++) {
  753. ret = svc_rdma_post_recv(newxprt);
  754. if (ret) {
  755. dprintk("svcrdma: failure posting receive buffers\n");
  756. goto errout;
  757. }
  758. }
  759. /* Swap out the handler */
  760. newxprt->sc_cm_id->event_handler = rdma_cma_handler;
  761. /* Accept Connection */
  762. set_bit(RDMAXPRT_CONN_PENDING, &newxprt->sc_flags);
  763. memset(&conn_param, 0, sizeof conn_param);
  764. conn_param.responder_resources = 0;
  765. conn_param.initiator_depth = newxprt->sc_ord;
  766. ret = rdma_accept(newxprt->sc_cm_id, &conn_param);
  767. if (ret) {
  768. dprintk("svcrdma: failed to accept new connection, ret=%d\n",
  769. ret);
  770. goto errout;
  771. }
  772. dprintk("svcrdma: new connection %p accepted with the following "
  773. "attributes:\n"
  774. " local_ip : %d.%d.%d.%d\n"
  775. " local_port : %d\n"
  776. " remote_ip : %d.%d.%d.%d\n"
  777. " remote_port : %d\n"
  778. " max_sge : %d\n"
  779. " sq_depth : %d\n"
  780. " max_requests : %d\n"
  781. " ord : %d\n",
  782. newxprt,
  783. NIPQUAD(((struct sockaddr_in *)&newxprt->sc_cm_id->
  784. route.addr.src_addr)->sin_addr.s_addr),
  785. ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
  786. route.addr.src_addr)->sin_port),
  787. NIPQUAD(((struct sockaddr_in *)&newxprt->sc_cm_id->
  788. route.addr.dst_addr)->sin_addr.s_addr),
  789. ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
  790. route.addr.dst_addr)->sin_port),
  791. newxprt->sc_max_sge,
  792. newxprt->sc_sq_depth,
  793. newxprt->sc_max_requests,
  794. newxprt->sc_ord);
  795. /* Set the local and remote addresses in the transport */
  796. sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
  797. svc_xprt_set_remote(&newxprt->sc_xprt, sa, svc_addr_len(sa));
  798. sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
  799. svc_xprt_set_local(&newxprt->sc_xprt, sa, svc_addr_len(sa));
  800. ib_req_notify_cq(newxprt->sc_sq_cq, IB_CQ_NEXT_COMP);
  801. ib_req_notify_cq(newxprt->sc_rq_cq, IB_CQ_NEXT_COMP);
  802. return &newxprt->sc_xprt;
  803. errout:
  804. dprintk("svcrdma: failure accepting new connection rc=%d.\n", ret);
  805. /* Take a reference in case the DTO handler runs */
  806. svc_xprt_get(&newxprt->sc_xprt);
  807. if (newxprt->sc_qp && !IS_ERR(newxprt->sc_qp)) {
  808. ib_destroy_qp(newxprt->sc_qp);
  809. svc_xprt_put(&newxprt->sc_xprt);
  810. }
  811. rdma_destroy_id(newxprt->sc_cm_id);
  812. /* This call to put will destroy the transport */
  813. svc_xprt_put(&newxprt->sc_xprt);
  814. return NULL;
  815. }
  816. static void svc_rdma_release_rqst(struct svc_rqst *rqstp)
  817. {
  818. }
  819. /*
  820. * When connected, an svc_xprt has at least three references:
  821. *
  822. * - A reference held by the QP. We still hold that here because this
  823. * code deletes the QP and puts the reference.
  824. *
  825. * - A reference held by the cm_id between the ESTABLISHED and
  826. * DISCONNECTED events. If the remote peer disconnected first, this
  827. * reference could be gone.
  828. *
  829. * - A reference held by the svc_recv code that called this function
  830. * as part of close processing.
  831. *
  832. * At a minimum two references should still be held.
  833. */
  834. static void svc_rdma_detach(struct svc_xprt *xprt)
  835. {
  836. struct svcxprt_rdma *rdma =
  837. container_of(xprt, struct svcxprt_rdma, sc_xprt);
  838. dprintk("svc: svc_rdma_detach(%p)\n", xprt);
  839. /* Disconnect and flush posted WQE */
  840. rdma_disconnect(rdma->sc_cm_id);
  841. /* Destroy the QP if present (not a listener) */
  842. if (rdma->sc_qp && !IS_ERR(rdma->sc_qp)) {
  843. ib_destroy_qp(rdma->sc_qp);
  844. svc_xprt_put(xprt);
  845. }
  846. /* Destroy the CM ID */
  847. rdma_destroy_id(rdma->sc_cm_id);
  848. }
  849. static void svc_rdma_free(struct svc_xprt *xprt)
  850. {
  851. struct svcxprt_rdma *rdma = (struct svcxprt_rdma *)xprt;
  852. dprintk("svcrdma: svc_rdma_free(%p)\n", rdma);
  853. /* We should only be called from kref_put */
  854. BUG_ON(atomic_read(&xprt->xpt_ref.refcount) != 0);
  855. if (rdma->sc_sq_cq && !IS_ERR(rdma->sc_sq_cq))
  856. ib_destroy_cq(rdma->sc_sq_cq);
  857. if (rdma->sc_rq_cq && !IS_ERR(rdma->sc_rq_cq))
  858. ib_destroy_cq(rdma->sc_rq_cq);
  859. if (rdma->sc_phys_mr && !IS_ERR(rdma->sc_phys_mr))
  860. ib_dereg_mr(rdma->sc_phys_mr);
  861. if (rdma->sc_pd && !IS_ERR(rdma->sc_pd))
  862. ib_dealloc_pd(rdma->sc_pd);
  863. destroy_context_cache(rdma->sc_ctxt_head);
  864. kfree(rdma);
  865. }
  866. static int svc_rdma_has_wspace(struct svc_xprt *xprt)
  867. {
  868. struct svcxprt_rdma *rdma =
  869. container_of(xprt, struct svcxprt_rdma, sc_xprt);
  870. /*
  871. * If there are fewer SQ WR available than required to send a
  872. * simple response, return false.
  873. */
  874. if ((rdma->sc_sq_depth - atomic_read(&rdma->sc_sq_count) < 3))
  875. return 0;
  876. /*
  877. * ...or there are already waiters on the SQ,
  878. * return false.
  879. */
  880. if (waitqueue_active(&rdma->sc_send_wait))
  881. return 0;
  882. /* Otherwise return true. */
  883. return 1;
  884. }
  885. int svc_rdma_send(struct svcxprt_rdma *xprt, struct ib_send_wr *wr)
  886. {
  887. struct ib_send_wr *bad_wr;
  888. int ret;
  889. if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
  890. return -ENOTCONN;
  891. BUG_ON(wr->send_flags != IB_SEND_SIGNALED);
  892. BUG_ON(((struct svc_rdma_op_ctxt *)(unsigned long)wr->wr_id)->wr_op !=
  893. wr->opcode);
  894. /* If the SQ is full, wait until an SQ entry is available */
  895. while (1) {
  896. spin_lock_bh(&xprt->sc_lock);
  897. if (xprt->sc_sq_depth == atomic_read(&xprt->sc_sq_count)) {
  898. spin_unlock_bh(&xprt->sc_lock);
  899. atomic_inc(&rdma_stat_sq_starve);
  900. /* See if we can opportunistically reap SQ WR to make room */
  901. sq_cq_reap(xprt);
  902. /* Wait until SQ WR available if SQ still full */
  903. wait_event(xprt->sc_send_wait,
  904. atomic_read(&xprt->sc_sq_count) <
  905. xprt->sc_sq_depth);
  906. if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
  907. return 0;
  908. continue;
  909. }
  910. /* Bumped used SQ WR count and post */
  911. ret = ib_post_send(xprt->sc_qp, wr, &bad_wr);
  912. if (!ret)
  913. atomic_inc(&xprt->sc_sq_count);
  914. else
  915. dprintk("svcrdma: failed to post SQ WR rc=%d, "
  916. "sc_sq_count=%d, sc_sq_depth=%d\n",
  917. ret, atomic_read(&xprt->sc_sq_count),
  918. xprt->sc_sq_depth);
  919. spin_unlock_bh(&xprt->sc_lock);
  920. break;
  921. }
  922. return ret;
  923. }
  924. int svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp,
  925. enum rpcrdma_errcode err)
  926. {
  927. struct ib_send_wr err_wr;
  928. struct ib_sge sge;
  929. struct page *p;
  930. struct svc_rdma_op_ctxt *ctxt;
  931. u32 *va;
  932. int length;
  933. int ret;
  934. p = svc_rdma_get_page();
  935. va = page_address(p);
  936. /* XDR encode error */
  937. length = svc_rdma_xdr_encode_error(xprt, rmsgp, err, va);
  938. /* Prepare SGE for local address */
  939. sge.addr = ib_dma_map_page(xprt->sc_cm_id->device,
  940. p, 0, PAGE_SIZE, DMA_FROM_DEVICE);
  941. sge.lkey = xprt->sc_phys_mr->lkey;
  942. sge.length = length;
  943. ctxt = svc_rdma_get_context(xprt);
  944. ctxt->count = 1;
  945. ctxt->pages[0] = p;
  946. /* Prepare SEND WR */
  947. memset(&err_wr, 0, sizeof err_wr);
  948. ctxt->wr_op = IB_WR_SEND;
  949. err_wr.wr_id = (unsigned long)ctxt;
  950. err_wr.sg_list = &sge;
  951. err_wr.num_sge = 1;
  952. err_wr.opcode = IB_WR_SEND;
  953. err_wr.send_flags = IB_SEND_SIGNALED;
  954. /* Post It */
  955. ret = svc_rdma_send(xprt, &err_wr);
  956. if (ret) {
  957. dprintk("svcrdma: Error posting send = %d\n", ret);
  958. svc_rdma_put_context(ctxt, 1);
  959. }
  960. return ret;
  961. }