ar-recvmsg.c 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444
  1. /* RxRPC recvmsg() implementation
  2. *
  3. * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
  4. * Written by David Howells (dhowells@redhat.com)
  5. *
  6. * This program is free software; you can redistribute it and/or
  7. * modify it under the terms of the GNU General Public License
  8. * as published by the Free Software Foundation; either version
  9. * 2 of the License, or (at your option) any later version.
  10. */
  11. #include <linux/net.h>
  12. #include <linux/skbuff.h>
  13. #include <linux/export.h>
  14. #include <net/sock.h>
  15. #include <net/af_rxrpc.h>
  16. #include "ar-internal.h"
  17. /*
  18. * removal a call's user ID from the socket tree to make the user ID available
  19. * again and so that it won't be seen again in association with that call
  20. */
  21. void rxrpc_remove_user_ID(struct rxrpc_sock *rx, struct rxrpc_call *call)
  22. {
  23. _debug("RELEASE CALL %d", call->debug_id);
  24. if (test_bit(RXRPC_CALL_HAS_USERID, &call->flags)) {
  25. write_lock_bh(&rx->call_lock);
  26. rb_erase(&call->sock_node, &call->socket->calls);
  27. clear_bit(RXRPC_CALL_HAS_USERID, &call->flags);
  28. write_unlock_bh(&rx->call_lock);
  29. }
  30. read_lock_bh(&call->state_lock);
  31. if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
  32. !test_and_set_bit(RXRPC_CALL_RELEASE, &call->events))
  33. rxrpc_queue_call(call);
  34. read_unlock_bh(&call->state_lock);
  35. }
  36. /*
  37. * receive a message from an RxRPC socket
  38. * - we need to be careful about two or more threads calling recvmsg
  39. * simultaneously
  40. */
  41. int rxrpc_recvmsg(struct kiocb *iocb, struct socket *sock,
  42. struct msghdr *msg, size_t len, int flags)
  43. {
  44. struct rxrpc_skb_priv *sp;
  45. struct rxrpc_call *call = NULL, *continue_call = NULL;
  46. struct rxrpc_sock *rx = rxrpc_sk(sock->sk);
  47. struct sk_buff *skb;
  48. long timeo;
  49. int copy, ret, ullen, offset, copied = 0;
  50. u32 abort_code;
  51. DEFINE_WAIT(wait);
  52. _enter(",,,%zu,%d", len, flags);
  53. if (flags & (MSG_OOB | MSG_TRUNC))
  54. return -EOPNOTSUPP;
  55. ullen = msg->msg_flags & MSG_CMSG_COMPAT ? 4 : sizeof(unsigned long);
  56. timeo = sock_rcvtimeo(&rx->sk, flags & MSG_DONTWAIT);
  57. msg->msg_flags |= MSG_MORE;
  58. lock_sock(&rx->sk);
  59. for (;;) {
  60. /* return immediately if a client socket has no outstanding
  61. * calls */
  62. if (RB_EMPTY_ROOT(&rx->calls)) {
  63. if (copied)
  64. goto out;
  65. if (rx->sk.sk_state != RXRPC_SERVER_LISTENING) {
  66. release_sock(&rx->sk);
  67. if (continue_call)
  68. rxrpc_put_call(continue_call);
  69. return -ENODATA;
  70. }
  71. }
  72. /* get the next message on the Rx queue */
  73. skb = skb_peek(&rx->sk.sk_receive_queue);
  74. if (!skb) {
  75. /* nothing remains on the queue */
  76. if (copied &&
  77. (msg->msg_flags & MSG_PEEK || timeo == 0))
  78. goto out;
  79. /* wait for a message to turn up */
  80. release_sock(&rx->sk);
  81. prepare_to_wait_exclusive(sk_sleep(&rx->sk), &wait,
  82. TASK_INTERRUPTIBLE);
  83. ret = sock_error(&rx->sk);
  84. if (ret)
  85. goto wait_error;
  86. if (skb_queue_empty(&rx->sk.sk_receive_queue)) {
  87. if (signal_pending(current))
  88. goto wait_interrupted;
  89. timeo = schedule_timeout(timeo);
  90. }
  91. finish_wait(sk_sleep(&rx->sk), &wait);
  92. lock_sock(&rx->sk);
  93. continue;
  94. }
  95. peek_next_packet:
  96. sp = rxrpc_skb(skb);
  97. call = sp->call;
  98. ASSERT(call != NULL);
  99. _debug("next pkt %s", rxrpc_pkts[sp->hdr.type]);
  100. /* make sure we wait for the state to be updated in this call */
  101. spin_lock_bh(&call->lock);
  102. spin_unlock_bh(&call->lock);
  103. if (test_bit(RXRPC_CALL_RELEASED, &call->flags)) {
  104. _debug("packet from released call");
  105. if (skb_dequeue(&rx->sk.sk_receive_queue) != skb)
  106. BUG();
  107. rxrpc_free_skb(skb);
  108. continue;
  109. }
  110. /* determine whether to continue last data receive */
  111. if (continue_call) {
  112. _debug("maybe cont");
  113. if (call != continue_call ||
  114. skb->mark != RXRPC_SKB_MARK_DATA) {
  115. release_sock(&rx->sk);
  116. rxrpc_put_call(continue_call);
  117. _leave(" = %d [noncont]", copied);
  118. return copied;
  119. }
  120. }
  121. rxrpc_get_call(call);
  122. /* copy the peer address and timestamp */
  123. if (!continue_call) {
  124. if (msg->msg_name) {
  125. size_t len =
  126. sizeof(call->conn->trans->peer->srx);
  127. memcpy(msg->msg_name,
  128. &call->conn->trans->peer->srx, len);
  129. msg->msg_namelen = len;
  130. }
  131. sock_recv_ts_and_drops(msg, &rx->sk, skb);
  132. }
  133. /* receive the message */
  134. if (skb->mark != RXRPC_SKB_MARK_DATA)
  135. goto receive_non_data_message;
  136. _debug("recvmsg DATA #%u { %d, %d }",
  137. ntohl(sp->hdr.seq), skb->len, sp->offset);
  138. if (!continue_call) {
  139. /* only set the control data once per recvmsg() */
  140. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID,
  141. ullen, &call->user_call_ID);
  142. if (ret < 0)
  143. goto copy_error;
  144. ASSERT(test_bit(RXRPC_CALL_HAS_USERID, &call->flags));
  145. }
  146. ASSERTCMP(ntohl(sp->hdr.seq), >=, call->rx_data_recv);
  147. ASSERTCMP(ntohl(sp->hdr.seq), <=, call->rx_data_recv + 1);
  148. call->rx_data_recv = ntohl(sp->hdr.seq);
  149. ASSERTCMP(ntohl(sp->hdr.seq), >, call->rx_data_eaten);
  150. offset = sp->offset;
  151. copy = skb->len - offset;
  152. if (copy > len - copied)
  153. copy = len - copied;
  154. if (skb->ip_summed == CHECKSUM_UNNECESSARY) {
  155. ret = skb_copy_datagram_iovec(skb, offset,
  156. msg->msg_iov, copy);
  157. } else {
  158. ret = skb_copy_and_csum_datagram_iovec(skb, offset,
  159. msg->msg_iov,
  160. copy);
  161. if (ret == -EINVAL)
  162. goto csum_copy_error;
  163. }
  164. if (ret < 0)
  165. goto copy_error;
  166. /* handle piecemeal consumption of data packets */
  167. _debug("copied %d+%d", copy, copied);
  168. offset += copy;
  169. copied += copy;
  170. if (!(flags & MSG_PEEK))
  171. sp->offset = offset;
  172. if (sp->offset < skb->len) {
  173. _debug("buffer full");
  174. ASSERTCMP(copied, ==, len);
  175. break;
  176. }
  177. /* we transferred the whole data packet */
  178. if (sp->hdr.flags & RXRPC_LAST_PACKET) {
  179. _debug("last");
  180. if (call->conn->out_clientflag) {
  181. /* last byte of reply received */
  182. ret = copied;
  183. goto terminal_message;
  184. }
  185. /* last bit of request received */
  186. if (!(flags & MSG_PEEK)) {
  187. _debug("eat packet");
  188. if (skb_dequeue(&rx->sk.sk_receive_queue) !=
  189. skb)
  190. BUG();
  191. rxrpc_free_skb(skb);
  192. }
  193. msg->msg_flags &= ~MSG_MORE;
  194. break;
  195. }
  196. /* move on to the next data message */
  197. _debug("next");
  198. if (!continue_call)
  199. continue_call = sp->call;
  200. else
  201. rxrpc_put_call(call);
  202. call = NULL;
  203. if (flags & MSG_PEEK) {
  204. _debug("peek next");
  205. skb = skb->next;
  206. if (skb == (struct sk_buff *) &rx->sk.sk_receive_queue)
  207. break;
  208. goto peek_next_packet;
  209. }
  210. _debug("eat packet");
  211. if (skb_dequeue(&rx->sk.sk_receive_queue) != skb)
  212. BUG();
  213. rxrpc_free_skb(skb);
  214. }
  215. /* end of non-terminal data packet reception for the moment */
  216. _debug("end rcv data");
  217. out:
  218. release_sock(&rx->sk);
  219. if (call)
  220. rxrpc_put_call(call);
  221. if (continue_call)
  222. rxrpc_put_call(continue_call);
  223. _leave(" = %d [data]", copied);
  224. return copied;
  225. /* handle non-DATA messages such as aborts, incoming connections and
  226. * final ACKs */
  227. receive_non_data_message:
  228. _debug("non-data");
  229. if (skb->mark == RXRPC_SKB_MARK_NEW_CALL) {
  230. _debug("RECV NEW CALL");
  231. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_NEW_CALL, 0, &abort_code);
  232. if (ret < 0)
  233. goto copy_error;
  234. if (!(flags & MSG_PEEK)) {
  235. if (skb_dequeue(&rx->sk.sk_receive_queue) != skb)
  236. BUG();
  237. rxrpc_free_skb(skb);
  238. }
  239. goto out;
  240. }
  241. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID,
  242. ullen, &call->user_call_ID);
  243. if (ret < 0)
  244. goto copy_error;
  245. ASSERT(test_bit(RXRPC_CALL_HAS_USERID, &call->flags));
  246. switch (skb->mark) {
  247. case RXRPC_SKB_MARK_DATA:
  248. BUG();
  249. case RXRPC_SKB_MARK_FINAL_ACK:
  250. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ACK, 0, &abort_code);
  251. break;
  252. case RXRPC_SKB_MARK_BUSY:
  253. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_BUSY, 0, &abort_code);
  254. break;
  255. case RXRPC_SKB_MARK_REMOTE_ABORT:
  256. abort_code = call->abort_code;
  257. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &abort_code);
  258. break;
  259. case RXRPC_SKB_MARK_NET_ERROR:
  260. _debug("RECV NET ERROR %d", sp->error);
  261. abort_code = sp->error;
  262. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_NET_ERROR, 4, &abort_code);
  263. break;
  264. case RXRPC_SKB_MARK_LOCAL_ERROR:
  265. _debug("RECV LOCAL ERROR %d", sp->error);
  266. abort_code = sp->error;
  267. ret = put_cmsg(msg, SOL_RXRPC, RXRPC_LOCAL_ERROR, 4,
  268. &abort_code);
  269. break;
  270. default:
  271. BUG();
  272. break;
  273. }
  274. if (ret < 0)
  275. goto copy_error;
  276. terminal_message:
  277. _debug("terminal");
  278. msg->msg_flags &= ~MSG_MORE;
  279. msg->msg_flags |= MSG_EOR;
  280. if (!(flags & MSG_PEEK)) {
  281. _net("free terminal skb %p", skb);
  282. if (skb_dequeue(&rx->sk.sk_receive_queue) != skb)
  283. BUG();
  284. rxrpc_free_skb(skb);
  285. rxrpc_remove_user_ID(rx, call);
  286. }
  287. release_sock(&rx->sk);
  288. rxrpc_put_call(call);
  289. if (continue_call)
  290. rxrpc_put_call(continue_call);
  291. _leave(" = %d", ret);
  292. return ret;
  293. copy_error:
  294. _debug("copy error");
  295. release_sock(&rx->sk);
  296. rxrpc_put_call(call);
  297. if (continue_call)
  298. rxrpc_put_call(continue_call);
  299. _leave(" = %d", ret);
  300. return ret;
  301. csum_copy_error:
  302. _debug("csum error");
  303. release_sock(&rx->sk);
  304. if (continue_call)
  305. rxrpc_put_call(continue_call);
  306. rxrpc_kill_skb(skb);
  307. skb_kill_datagram(&rx->sk, skb, flags);
  308. rxrpc_put_call(call);
  309. return -EAGAIN;
  310. wait_interrupted:
  311. ret = sock_intr_errno(timeo);
  312. wait_error:
  313. finish_wait(sk_sleep(&rx->sk), &wait);
  314. if (continue_call)
  315. rxrpc_put_call(continue_call);
  316. if (copied)
  317. copied = ret;
  318. _leave(" = %d [waitfail %d]", copied, ret);
  319. return copied;
  320. }
  321. /**
  322. * rxrpc_kernel_data_delivered - Record delivery of data message
  323. * @skb: Message holding data
  324. *
  325. * Record the delivery of a data message. This permits RxRPC to keep its
  326. * tracking correct. The socket buffer will be deleted.
  327. */
  328. void rxrpc_kernel_data_delivered(struct sk_buff *skb)
  329. {
  330. struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  331. struct rxrpc_call *call = sp->call;
  332. ASSERTCMP(ntohl(sp->hdr.seq), >=, call->rx_data_recv);
  333. ASSERTCMP(ntohl(sp->hdr.seq), <=, call->rx_data_recv + 1);
  334. call->rx_data_recv = ntohl(sp->hdr.seq);
  335. ASSERTCMP(ntohl(sp->hdr.seq), >, call->rx_data_eaten);
  336. rxrpc_free_skb(skb);
  337. }
  338. EXPORT_SYMBOL(rxrpc_kernel_data_delivered);
  339. /**
  340. * rxrpc_kernel_is_data_last - Determine if data message is last one
  341. * @skb: Message holding data
  342. *
  343. * Determine if data message is last one for the parent call.
  344. */
  345. bool rxrpc_kernel_is_data_last(struct sk_buff *skb)
  346. {
  347. struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  348. ASSERTCMP(skb->mark, ==, RXRPC_SKB_MARK_DATA);
  349. return sp->hdr.flags & RXRPC_LAST_PACKET;
  350. }
  351. EXPORT_SYMBOL(rxrpc_kernel_is_data_last);
  352. /**
  353. * rxrpc_kernel_get_abort_code - Get the abort code from an RxRPC abort message
  354. * @skb: Message indicating an abort
  355. *
  356. * Get the abort code from an RxRPC abort message.
  357. */
  358. u32 rxrpc_kernel_get_abort_code(struct sk_buff *skb)
  359. {
  360. struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  361. ASSERTCMP(skb->mark, ==, RXRPC_SKB_MARK_REMOTE_ABORT);
  362. return sp->call->abort_code;
  363. }
  364. EXPORT_SYMBOL(rxrpc_kernel_get_abort_code);
  365. /**
  366. * rxrpc_kernel_get_error - Get the error number from an RxRPC error message
  367. * @skb: Message indicating an error
  368. *
  369. * Get the error number from an RxRPC error message.
  370. */
  371. int rxrpc_kernel_get_error_number(struct sk_buff *skb)
  372. {
  373. struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  374. return sp->error;
  375. }
  376. EXPORT_SYMBOL(rxrpc_kernel_get_error_number);