recvmsg.c 10 KB

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