smc_rx.c 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444
  1. // SPDX-License-Identifier: GPL-2.0
  2. /*
  3. * Shared Memory Communications over RDMA (SMC-R) and RoCE
  4. *
  5. * Manage RMBE
  6. * copy new RMBE data into user space
  7. *
  8. * Copyright IBM Corp. 2016
  9. *
  10. * Author(s): Ursula Braun <ubraun@linux.vnet.ibm.com>
  11. */
  12. #include <linux/net.h>
  13. #include <linux/rcupdate.h>
  14. #include <linux/sched/signal.h>
  15. #include <net/sock.h>
  16. #include "smc.h"
  17. #include "smc_core.h"
  18. #include "smc_cdc.h"
  19. #include "smc_tx.h" /* smc_tx_consumer_update() */
  20. #include "smc_rx.h"
  21. /* callback implementation to wakeup consumers blocked with smc_rx_wait().
  22. * indirectly called by smc_cdc_msg_recv_action().
  23. */
  24. static void smc_rx_wake_up(struct sock *sk)
  25. {
  26. struct socket_wq *wq;
  27. /* derived from sock_def_readable() */
  28. /* called already in smc_listen_work() */
  29. rcu_read_lock();
  30. wq = rcu_dereference(sk->sk_wq);
  31. if (skwq_has_sleeper(wq))
  32. wake_up_interruptible_sync_poll(&wq->wait, EPOLLIN | EPOLLPRI |
  33. EPOLLRDNORM | EPOLLRDBAND);
  34. sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
  35. if ((sk->sk_shutdown == SHUTDOWN_MASK) ||
  36. (sk->sk_state == SMC_CLOSED))
  37. sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_HUP);
  38. rcu_read_unlock();
  39. }
  40. /* Update consumer cursor
  41. * @conn connection to update
  42. * @cons consumer cursor
  43. * @len number of Bytes consumed
  44. * Returns:
  45. * 1 if we should end our receive, 0 otherwise
  46. */
  47. static int smc_rx_update_consumer(struct smc_sock *smc,
  48. union smc_host_cursor cons, size_t len)
  49. {
  50. struct smc_connection *conn = &smc->conn;
  51. struct sock *sk = &smc->sk;
  52. bool force = false;
  53. int diff, rc = 0;
  54. smc_curs_add(conn->rmb_desc->len, &cons, len);
  55. /* did we process urgent data? */
  56. if (conn->urg_state == SMC_URG_VALID || conn->urg_rx_skip_pend) {
  57. diff = smc_curs_comp(conn->rmb_desc->len, &cons,
  58. &conn->urg_curs);
  59. if (sock_flag(sk, SOCK_URGINLINE)) {
  60. if (diff == 0) {
  61. force = true;
  62. rc = 1;
  63. conn->urg_state = SMC_URG_READ;
  64. }
  65. } else {
  66. if (diff == 1) {
  67. /* skip urgent byte */
  68. force = true;
  69. smc_curs_add(conn->rmb_desc->len, &cons, 1);
  70. conn->urg_rx_skip_pend = false;
  71. } else if (diff < -1)
  72. /* we read past urgent byte */
  73. conn->urg_state = SMC_URG_READ;
  74. }
  75. }
  76. smc_curs_copy(&conn->local_tx_ctrl.cons, &cons, conn);
  77. /* send consumer cursor update if required */
  78. /* similar to advertising new TCP rcv_wnd if required */
  79. smc_tx_consumer_update(conn, force);
  80. return rc;
  81. }
  82. static void smc_rx_update_cons(struct smc_sock *smc, size_t len)
  83. {
  84. struct smc_connection *conn = &smc->conn;
  85. union smc_host_cursor cons;
  86. smc_curs_copy(&cons, &conn->local_tx_ctrl.cons, conn);
  87. smc_rx_update_consumer(smc, cons, len);
  88. }
  89. struct smc_spd_priv {
  90. struct smc_sock *smc;
  91. size_t len;
  92. };
  93. static void smc_rx_pipe_buf_release(struct pipe_inode_info *pipe,
  94. struct pipe_buffer *buf)
  95. {
  96. struct smc_spd_priv *priv = (struct smc_spd_priv *)buf->private;
  97. struct smc_sock *smc = priv->smc;
  98. struct smc_connection *conn;
  99. struct sock *sk = &smc->sk;
  100. if (sk->sk_state == SMC_CLOSED ||
  101. sk->sk_state == SMC_PEERFINCLOSEWAIT ||
  102. sk->sk_state == SMC_APPFINCLOSEWAIT)
  103. goto out;
  104. conn = &smc->conn;
  105. lock_sock(sk);
  106. smc_rx_update_cons(smc, priv->len);
  107. release_sock(sk);
  108. if (atomic_sub_and_test(priv->len, &conn->splice_pending))
  109. smc_rx_wake_up(sk);
  110. out:
  111. kfree(priv);
  112. put_page(buf->page);
  113. sock_put(sk);
  114. }
  115. static const struct pipe_buf_operations smc_pipe_ops = {
  116. .release = smc_rx_pipe_buf_release,
  117. .get = generic_pipe_buf_get
  118. };
  119. static void smc_rx_spd_release(struct splice_pipe_desc *spd,
  120. unsigned int i)
  121. {
  122. put_page(spd->pages[i]);
  123. }
  124. static int smc_rx_splice(struct pipe_inode_info *pipe, char *src, size_t len,
  125. struct smc_sock *smc)
  126. {
  127. struct splice_pipe_desc spd;
  128. struct partial_page partial;
  129. struct smc_spd_priv *priv;
  130. int bytes;
  131. priv = kzalloc(sizeof(*priv), GFP_KERNEL);
  132. if (!priv)
  133. return -ENOMEM;
  134. priv->len = len;
  135. priv->smc = smc;
  136. partial.offset = src - (char *)smc->conn.rmb_desc->cpu_addr;
  137. partial.len = len;
  138. partial.private = (unsigned long)priv;
  139. spd.nr_pages_max = 1;
  140. spd.nr_pages = 1;
  141. spd.pages = &smc->conn.rmb_desc->pages;
  142. spd.partial = &partial;
  143. spd.ops = &smc_pipe_ops;
  144. spd.spd_release = smc_rx_spd_release;
  145. bytes = splice_to_pipe(pipe, &spd);
  146. if (bytes > 0) {
  147. sock_hold(&smc->sk);
  148. get_page(smc->conn.rmb_desc->pages);
  149. atomic_add(bytes, &smc->conn.splice_pending);
  150. }
  151. return bytes;
  152. }
  153. static int smc_rx_data_available_and_no_splice_pend(struct smc_connection *conn)
  154. {
  155. return atomic_read(&conn->bytes_to_rcv) &&
  156. !atomic_read(&conn->splice_pending);
  157. }
  158. /* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
  159. * @smc smc socket
  160. * @timeo pointer to max seconds to wait, pointer to value 0 for no timeout
  161. * @fcrit add'l criterion to evaluate as function pointer
  162. * Returns:
  163. * 1 if at least 1 byte available in rcvbuf or if socket error/shutdown.
  164. * 0 otherwise (nothing in rcvbuf nor timeout, e.g. interrupted).
  165. */
  166. int smc_rx_wait(struct smc_sock *smc, long *timeo,
  167. int (*fcrit)(struct smc_connection *conn))
  168. {
  169. DEFINE_WAIT_FUNC(wait, woken_wake_function);
  170. struct smc_connection *conn = &smc->conn;
  171. struct smc_cdc_conn_state_flags *cflags =
  172. &conn->local_tx_ctrl.conn_state_flags;
  173. struct sock *sk = &smc->sk;
  174. int rc;
  175. if (fcrit(conn))
  176. return 1;
  177. sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
  178. add_wait_queue(sk_sleep(sk), &wait);
  179. rc = sk_wait_event(sk, timeo,
  180. sk->sk_err ||
  181. cflags->peer_conn_abort ||
  182. sk->sk_shutdown & RCV_SHUTDOWN ||
  183. conn->killed ||
  184. fcrit(conn),
  185. &wait);
  186. remove_wait_queue(sk_sleep(sk), &wait);
  187. sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
  188. return rc;
  189. }
  190. static int smc_rx_recv_urg(struct smc_sock *smc, struct msghdr *msg, int len,
  191. int flags)
  192. {
  193. struct smc_connection *conn = &smc->conn;
  194. union smc_host_cursor cons;
  195. struct sock *sk = &smc->sk;
  196. int rc = 0;
  197. if (sock_flag(sk, SOCK_URGINLINE) ||
  198. !(conn->urg_state == SMC_URG_VALID) ||
  199. conn->urg_state == SMC_URG_READ)
  200. return -EINVAL;
  201. if (conn->urg_state == SMC_URG_VALID) {
  202. if (!(flags & MSG_PEEK))
  203. smc->conn.urg_state = SMC_URG_READ;
  204. msg->msg_flags |= MSG_OOB;
  205. if (len > 0) {
  206. if (!(flags & MSG_TRUNC))
  207. rc = memcpy_to_msg(msg, &conn->urg_rx_byte, 1);
  208. len = 1;
  209. smc_curs_copy(&cons, &conn->local_tx_ctrl.cons, conn);
  210. if (smc_curs_diff(conn->rmb_desc->len, &cons,
  211. &conn->urg_curs) > 1)
  212. conn->urg_rx_skip_pend = true;
  213. /* Urgent Byte was already accounted for, but trigger
  214. * skipping the urgent byte in non-inline case
  215. */
  216. if (!(flags & MSG_PEEK))
  217. smc_rx_update_consumer(smc, cons, 0);
  218. } else {
  219. msg->msg_flags |= MSG_TRUNC;
  220. }
  221. return rc ? -EFAULT : len;
  222. }
  223. if (sk->sk_state == SMC_CLOSED || sk->sk_shutdown & RCV_SHUTDOWN)
  224. return 0;
  225. return -EAGAIN;
  226. }
  227. static bool smc_rx_recvmsg_data_available(struct smc_sock *smc)
  228. {
  229. struct smc_connection *conn = &smc->conn;
  230. if (smc_rx_data_available(conn))
  231. return true;
  232. else if (conn->urg_state == SMC_URG_VALID)
  233. /* we received a single urgent Byte - skip */
  234. smc_rx_update_cons(smc, 0);
  235. return false;
  236. }
  237. /* smc_rx_recvmsg - receive data from RMBE
  238. * @msg: copy data to receive buffer
  239. * @pipe: copy data to pipe if set - indicates splice() call
  240. *
  241. * rcvbuf consumer: main API called by socket layer.
  242. * Called under sk lock.
  243. */
  244. int smc_rx_recvmsg(struct smc_sock *smc, struct msghdr *msg,
  245. struct pipe_inode_info *pipe, size_t len, int flags)
  246. {
  247. size_t copylen, read_done = 0, read_remaining = len;
  248. size_t chunk_len, chunk_off, chunk_len_sum;
  249. struct smc_connection *conn = &smc->conn;
  250. int (*func)(struct smc_connection *conn);
  251. union smc_host_cursor cons;
  252. int readable, chunk;
  253. char *rcvbuf_base;
  254. struct sock *sk;
  255. int splbytes;
  256. long timeo;
  257. int target; /* Read at least these many bytes */
  258. int rc;
  259. if (unlikely(flags & MSG_ERRQUEUE))
  260. return -EINVAL; /* future work for sk.sk_family == AF_SMC */
  261. sk = &smc->sk;
  262. if (sk->sk_state == SMC_LISTEN)
  263. return -ENOTCONN;
  264. if (flags & MSG_OOB)
  265. return smc_rx_recv_urg(smc, msg, len, flags);
  266. timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
  267. target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
  268. /* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
  269. rcvbuf_base = conn->rx_off + conn->rmb_desc->cpu_addr;
  270. do { /* while (read_remaining) */
  271. if (read_done >= target || (pipe && read_done))
  272. break;
  273. if (conn->killed)
  274. break;
  275. if (smc_rx_recvmsg_data_available(smc))
  276. goto copy;
  277. if (sk->sk_shutdown & RCV_SHUTDOWN) {
  278. /* smc_cdc_msg_recv_action() could have run after
  279. * above smc_rx_recvmsg_data_available()
  280. */
  281. if (smc_rx_recvmsg_data_available(smc))
  282. goto copy;
  283. break;
  284. }
  285. if (read_done) {
  286. if (sk->sk_err ||
  287. sk->sk_state == SMC_CLOSED ||
  288. !timeo ||
  289. signal_pending(current))
  290. break;
  291. } else {
  292. if (sk->sk_err) {
  293. read_done = sock_error(sk);
  294. break;
  295. }
  296. if (sk->sk_state == SMC_CLOSED) {
  297. if (!sock_flag(sk, SOCK_DONE)) {
  298. /* This occurs when user tries to read
  299. * from never connected socket.
  300. */
  301. read_done = -ENOTCONN;
  302. break;
  303. }
  304. break;
  305. }
  306. if (signal_pending(current)) {
  307. read_done = sock_intr_errno(timeo);
  308. break;
  309. }
  310. if (!timeo)
  311. return -EAGAIN;
  312. }
  313. if (!smc_rx_data_available(conn)) {
  314. smc_rx_wait(smc, &timeo, smc_rx_data_available);
  315. continue;
  316. }
  317. copy:
  318. /* initialize variables for 1st iteration of subsequent loop */
  319. /* could be just 1 byte, even after waiting on data above */
  320. readable = atomic_read(&conn->bytes_to_rcv);
  321. splbytes = atomic_read(&conn->splice_pending);
  322. if (!readable || (msg && splbytes)) {
  323. if (splbytes)
  324. func = smc_rx_data_available_and_no_splice_pend;
  325. else
  326. func = smc_rx_data_available;
  327. smc_rx_wait(smc, &timeo, func);
  328. continue;
  329. }
  330. smc_curs_copy(&cons, &conn->local_tx_ctrl.cons, conn);
  331. /* subsequent splice() calls pick up where previous left */
  332. if (splbytes)
  333. smc_curs_add(conn->rmb_desc->len, &cons, splbytes);
  334. if (conn->urg_state == SMC_URG_VALID &&
  335. sock_flag(&smc->sk, SOCK_URGINLINE) &&
  336. readable > 1)
  337. readable--; /* always stop at urgent Byte */
  338. /* not more than what user space asked for */
  339. copylen = min_t(size_t, read_remaining, readable);
  340. /* determine chunks where to read from rcvbuf */
  341. /* either unwrapped case, or 1st chunk of wrapped case */
  342. chunk_len = min_t(size_t, copylen, conn->rmb_desc->len -
  343. cons.count);
  344. chunk_len_sum = chunk_len;
  345. chunk_off = cons.count;
  346. smc_rmb_sync_sg_for_cpu(conn);
  347. for (chunk = 0; chunk < 2; chunk++) {
  348. if (!(flags & MSG_TRUNC)) {
  349. if (msg) {
  350. rc = memcpy_to_msg(msg, rcvbuf_base +
  351. chunk_off,
  352. chunk_len);
  353. } else {
  354. rc = smc_rx_splice(pipe, rcvbuf_base +
  355. chunk_off, chunk_len,
  356. smc);
  357. }
  358. if (rc < 0) {
  359. if (!read_done)
  360. read_done = -EFAULT;
  361. smc_rmb_sync_sg_for_device(conn);
  362. goto out;
  363. }
  364. }
  365. read_remaining -= chunk_len;
  366. read_done += chunk_len;
  367. if (chunk_len_sum == copylen)
  368. break; /* either on 1st or 2nd iteration */
  369. /* prepare next (== 2nd) iteration */
  370. chunk_len = copylen - chunk_len; /* remainder */
  371. chunk_len_sum += chunk_len;
  372. chunk_off = 0; /* modulo offset in recv ring buffer */
  373. }
  374. smc_rmb_sync_sg_for_device(conn);
  375. /* update cursors */
  376. if (!(flags & MSG_PEEK)) {
  377. /* increased in recv tasklet smc_cdc_msg_rcv() */
  378. smp_mb__before_atomic();
  379. atomic_sub(copylen, &conn->bytes_to_rcv);
  380. /* guarantee 0 <= bytes_to_rcv <= rmb_desc->len */
  381. smp_mb__after_atomic();
  382. if (msg && smc_rx_update_consumer(smc, cons, copylen))
  383. goto out;
  384. }
  385. } while (read_remaining);
  386. out:
  387. return read_done;
  388. }
  389. /* Initialize receive properties on connection establishment. NB: not __init! */
  390. void smc_rx_init(struct smc_sock *smc)
  391. {
  392. smc->sk.sk_data_ready = smc_rx_wake_up;
  393. atomic_set(&smc->conn.splice_pending, 0);
  394. smc->conn.urg_state = SMC_URG_READ;
  395. }