rxrpc.c 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860
  1. /* Maintain an RxRPC server socket to do AFS communications through
  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/slab.h>
  12. #include <net/sock.h>
  13. #include <net/af_rxrpc.h>
  14. #include <rxrpc/packet.h>
  15. #include "internal.h"
  16. #include "afs_cm.h"
  17. static struct socket *afs_socket; /* my RxRPC socket */
  18. static struct workqueue_struct *afs_async_calls;
  19. static atomic_t afs_outstanding_calls;
  20. static atomic_t afs_outstanding_skbs;
  21. static void afs_wake_up_call_waiter(struct afs_call *);
  22. static int afs_wait_for_call_to_complete(struct afs_call *);
  23. static void afs_wake_up_async_call(struct afs_call *);
  24. static int afs_dont_wait_for_call_to_complete(struct afs_call *);
  25. static void afs_process_async_call(struct work_struct *);
  26. static void afs_rx_interceptor(struct sock *, unsigned long, struct sk_buff *);
  27. static int afs_deliver_cm_op_id(struct afs_call *, struct sk_buff *, bool);
  28. /* synchronous call management */
  29. const struct afs_wait_mode afs_sync_call = {
  30. .rx_wakeup = afs_wake_up_call_waiter,
  31. .wait = afs_wait_for_call_to_complete,
  32. };
  33. /* asynchronous call management */
  34. const struct afs_wait_mode afs_async_call = {
  35. .rx_wakeup = afs_wake_up_async_call,
  36. .wait = afs_dont_wait_for_call_to_complete,
  37. };
  38. /* asynchronous incoming call management */
  39. static const struct afs_wait_mode afs_async_incoming_call = {
  40. .rx_wakeup = afs_wake_up_async_call,
  41. };
  42. /* asynchronous incoming call initial processing */
  43. static const struct afs_call_type afs_RXCMxxxx = {
  44. .name = "CB.xxxx",
  45. .deliver = afs_deliver_cm_op_id,
  46. .abort_to_error = afs_abort_to_error,
  47. };
  48. static void afs_collect_incoming_call(struct work_struct *);
  49. static struct sk_buff_head afs_incoming_calls;
  50. static DECLARE_WORK(afs_collect_incoming_call_work, afs_collect_incoming_call);
  51. /*
  52. * open an RxRPC socket and bind it to be a server for callback notifications
  53. * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT
  54. */
  55. int afs_open_socket(void)
  56. {
  57. struct sockaddr_rxrpc srx;
  58. struct socket *socket;
  59. int ret;
  60. _enter("");
  61. skb_queue_head_init(&afs_incoming_calls);
  62. afs_async_calls = create_singlethread_workqueue("kafsd");
  63. if (!afs_async_calls) {
  64. _leave(" = -ENOMEM [wq]");
  65. return -ENOMEM;
  66. }
  67. ret = sock_create_kern(AF_RXRPC, SOCK_DGRAM, PF_INET, &socket);
  68. if (ret < 0) {
  69. destroy_workqueue(afs_async_calls);
  70. _leave(" = %d [socket]", ret);
  71. return ret;
  72. }
  73. socket->sk->sk_allocation = GFP_NOFS;
  74. /* bind the callback manager's address to make this a server socket */
  75. srx.srx_family = AF_RXRPC;
  76. srx.srx_service = CM_SERVICE;
  77. srx.transport_type = SOCK_DGRAM;
  78. srx.transport_len = sizeof(srx.transport.sin);
  79. srx.transport.sin.sin_family = AF_INET;
  80. srx.transport.sin.sin_port = htons(AFS_CM_PORT);
  81. memset(&srx.transport.sin.sin_addr, 0,
  82. sizeof(srx.transport.sin.sin_addr));
  83. ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx));
  84. if (ret < 0) {
  85. sock_release(socket);
  86. destroy_workqueue(afs_async_calls);
  87. _leave(" = %d [bind]", ret);
  88. return ret;
  89. }
  90. rxrpc_kernel_intercept_rx_messages(socket, afs_rx_interceptor);
  91. afs_socket = socket;
  92. _leave(" = 0");
  93. return 0;
  94. }
  95. /*
  96. * close the RxRPC socket AFS was using
  97. */
  98. void afs_close_socket(void)
  99. {
  100. _enter("");
  101. sock_release(afs_socket);
  102. _debug("dework");
  103. destroy_workqueue(afs_async_calls);
  104. ASSERTCMP(atomic_read(&afs_outstanding_skbs), ==, 0);
  105. ASSERTCMP(atomic_read(&afs_outstanding_calls), ==, 0);
  106. _leave("");
  107. }
  108. /*
  109. * note that the data in a socket buffer is now delivered and that the buffer
  110. * should be freed
  111. */
  112. static void afs_data_delivered(struct sk_buff *skb)
  113. {
  114. if (!skb) {
  115. _debug("DLVR NULL [%d]", atomic_read(&afs_outstanding_skbs));
  116. dump_stack();
  117. } else {
  118. _debug("DLVR %p{%u} [%d]",
  119. skb, skb->mark, atomic_read(&afs_outstanding_skbs));
  120. if (atomic_dec_return(&afs_outstanding_skbs) == -1)
  121. BUG();
  122. rxrpc_kernel_data_delivered(skb);
  123. }
  124. }
  125. /*
  126. * free a socket buffer
  127. */
  128. static void afs_free_skb(struct sk_buff *skb)
  129. {
  130. if (!skb) {
  131. _debug("FREE NULL [%d]", atomic_read(&afs_outstanding_skbs));
  132. dump_stack();
  133. } else {
  134. _debug("FREE %p{%u} [%d]",
  135. skb, skb->mark, atomic_read(&afs_outstanding_skbs));
  136. if (atomic_dec_return(&afs_outstanding_skbs) == -1)
  137. BUG();
  138. rxrpc_kernel_free_skb(skb);
  139. }
  140. }
  141. /*
  142. * free a call
  143. */
  144. static void afs_free_call(struct afs_call *call)
  145. {
  146. _debug("DONE %p{%s} [%d]",
  147. call, call->type->name, atomic_read(&afs_outstanding_calls));
  148. if (atomic_dec_return(&afs_outstanding_calls) == -1)
  149. BUG();
  150. ASSERTCMP(call->rxcall, ==, NULL);
  151. ASSERT(!work_pending(&call->async_work));
  152. ASSERT(skb_queue_empty(&call->rx_queue));
  153. ASSERT(call->type->name != NULL);
  154. kfree(call->request);
  155. kfree(call);
  156. }
  157. /*
  158. * allocate a call with flat request and reply buffers
  159. */
  160. struct afs_call *afs_alloc_flat_call(const struct afs_call_type *type,
  161. size_t request_size, size_t reply_size)
  162. {
  163. struct afs_call *call;
  164. call = kzalloc(sizeof(*call), GFP_NOFS);
  165. if (!call)
  166. goto nomem_call;
  167. _debug("CALL %p{%s} [%d]",
  168. call, type->name, atomic_read(&afs_outstanding_calls));
  169. atomic_inc(&afs_outstanding_calls);
  170. call->type = type;
  171. call->request_size = request_size;
  172. call->reply_max = reply_size;
  173. if (request_size) {
  174. call->request = kmalloc(request_size, GFP_NOFS);
  175. if (!call->request)
  176. goto nomem_free;
  177. }
  178. if (reply_size) {
  179. call->buffer = kmalloc(reply_size, GFP_NOFS);
  180. if (!call->buffer)
  181. goto nomem_free;
  182. }
  183. init_waitqueue_head(&call->waitq);
  184. skb_queue_head_init(&call->rx_queue);
  185. return call;
  186. nomem_free:
  187. afs_free_call(call);
  188. nomem_call:
  189. return NULL;
  190. }
  191. /*
  192. * clean up a call with flat buffer
  193. */
  194. void afs_flat_call_destructor(struct afs_call *call)
  195. {
  196. _enter("");
  197. kfree(call->request);
  198. call->request = NULL;
  199. kfree(call->buffer);
  200. call->buffer = NULL;
  201. }
  202. /*
  203. * attach the data from a bunch of pages on an inode to a call
  204. */
  205. static int afs_send_pages(struct afs_call *call, struct msghdr *msg,
  206. struct kvec *iov)
  207. {
  208. struct page *pages[8];
  209. unsigned count, n, loop, offset, to;
  210. pgoff_t first = call->first, last = call->last;
  211. int ret;
  212. _enter("");
  213. offset = call->first_offset;
  214. call->first_offset = 0;
  215. do {
  216. _debug("attach %lx-%lx", first, last);
  217. count = last - first + 1;
  218. if (count > ARRAY_SIZE(pages))
  219. count = ARRAY_SIZE(pages);
  220. n = find_get_pages_contig(call->mapping, first, count, pages);
  221. ASSERTCMP(n, ==, count);
  222. loop = 0;
  223. do {
  224. msg->msg_flags = 0;
  225. to = PAGE_SIZE;
  226. if (first + loop >= last)
  227. to = call->last_to;
  228. else
  229. msg->msg_flags = MSG_MORE;
  230. iov->iov_base = kmap(pages[loop]) + offset;
  231. iov->iov_len = to - offset;
  232. offset = 0;
  233. _debug("- range %u-%u%s",
  234. offset, to, msg->msg_flags ? " [more]" : "");
  235. msg->msg_iov = (struct iovec *) iov;
  236. msg->msg_iovlen = 1;
  237. /* have to change the state *before* sending the last
  238. * packet as RxRPC might give us the reply before it
  239. * returns from sending the request */
  240. if (first + loop >= last)
  241. call->state = AFS_CALL_AWAIT_REPLY;
  242. ret = rxrpc_kernel_send_data(call->rxcall, msg,
  243. to - offset);
  244. kunmap(pages[loop]);
  245. if (ret < 0)
  246. break;
  247. } while (++loop < count);
  248. first += count;
  249. for (loop = 0; loop < count; loop++)
  250. put_page(pages[loop]);
  251. if (ret < 0)
  252. break;
  253. } while (first <= last);
  254. _leave(" = %d", ret);
  255. return ret;
  256. }
  257. /*
  258. * initiate a call
  259. */
  260. int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
  261. const struct afs_wait_mode *wait_mode)
  262. {
  263. struct sockaddr_rxrpc srx;
  264. struct rxrpc_call *rxcall;
  265. struct msghdr msg;
  266. struct kvec iov[1];
  267. int ret;
  268. struct sk_buff *skb;
  269. _enter("%x,{%d},", addr->s_addr, ntohs(call->port));
  270. ASSERT(call->type != NULL);
  271. ASSERT(call->type->name != NULL);
  272. _debug("____MAKE %p{%s,%x} [%d]____",
  273. call, call->type->name, key_serial(call->key),
  274. atomic_read(&afs_outstanding_calls));
  275. call->wait_mode = wait_mode;
  276. INIT_WORK(&call->async_work, afs_process_async_call);
  277. memset(&srx, 0, sizeof(srx));
  278. srx.srx_family = AF_RXRPC;
  279. srx.srx_service = call->service_id;
  280. srx.transport_type = SOCK_DGRAM;
  281. srx.transport_len = sizeof(srx.transport.sin);
  282. srx.transport.sin.sin_family = AF_INET;
  283. srx.transport.sin.sin_port = call->port;
  284. memcpy(&srx.transport.sin.sin_addr, addr, 4);
  285. /* create a call */
  286. rxcall = rxrpc_kernel_begin_call(afs_socket, &srx, call->key,
  287. (unsigned long) call, gfp);
  288. call->key = NULL;
  289. if (IS_ERR(rxcall)) {
  290. ret = PTR_ERR(rxcall);
  291. goto error_kill_call;
  292. }
  293. call->rxcall = rxcall;
  294. /* send the request */
  295. iov[0].iov_base = call->request;
  296. iov[0].iov_len = call->request_size;
  297. msg.msg_name = NULL;
  298. msg.msg_namelen = 0;
  299. msg.msg_iov = (struct iovec *) iov;
  300. msg.msg_iovlen = 1;
  301. msg.msg_control = NULL;
  302. msg.msg_controllen = 0;
  303. msg.msg_flags = (call->send_pages ? MSG_MORE : 0);
  304. /* have to change the state *before* sending the last packet as RxRPC
  305. * might give us the reply before it returns from sending the
  306. * request */
  307. if (!call->send_pages)
  308. call->state = AFS_CALL_AWAIT_REPLY;
  309. ret = rxrpc_kernel_send_data(rxcall, &msg, call->request_size);
  310. if (ret < 0)
  311. goto error_do_abort;
  312. if (call->send_pages) {
  313. ret = afs_send_pages(call, &msg, iov);
  314. if (ret < 0)
  315. goto error_do_abort;
  316. }
  317. /* at this point, an async call may no longer exist as it may have
  318. * already completed */
  319. return wait_mode->wait(call);
  320. error_do_abort:
  321. rxrpc_kernel_abort_call(rxcall, RX_USER_ABORT);
  322. while ((skb = skb_dequeue(&call->rx_queue)))
  323. afs_free_skb(skb);
  324. rxrpc_kernel_end_call(rxcall);
  325. call->rxcall = NULL;
  326. error_kill_call:
  327. call->type->destructor(call);
  328. afs_free_call(call);
  329. _leave(" = %d", ret);
  330. return ret;
  331. }
  332. /*
  333. * handles intercepted messages that were arriving in the socket's Rx queue
  334. * - called with the socket receive queue lock held to ensure message ordering
  335. * - called with softirqs disabled
  336. */
  337. static void afs_rx_interceptor(struct sock *sk, unsigned long user_call_ID,
  338. struct sk_buff *skb)
  339. {
  340. struct afs_call *call = (struct afs_call *) user_call_ID;
  341. _enter("%p,,%u", call, skb->mark);
  342. _debug("ICPT %p{%u} [%d]",
  343. skb, skb->mark, atomic_read(&afs_outstanding_skbs));
  344. ASSERTCMP(sk, ==, afs_socket->sk);
  345. atomic_inc(&afs_outstanding_skbs);
  346. if (!call) {
  347. /* its an incoming call for our callback service */
  348. skb_queue_tail(&afs_incoming_calls, skb);
  349. queue_work(afs_wq, &afs_collect_incoming_call_work);
  350. } else {
  351. /* route the messages directly to the appropriate call */
  352. skb_queue_tail(&call->rx_queue, skb);
  353. call->wait_mode->rx_wakeup(call);
  354. }
  355. _leave("");
  356. }
  357. /*
  358. * deliver messages to a call
  359. */
  360. static void afs_deliver_to_call(struct afs_call *call)
  361. {
  362. struct sk_buff *skb;
  363. bool last;
  364. u32 abort_code;
  365. int ret;
  366. _enter("");
  367. while ((call->state == AFS_CALL_AWAIT_REPLY ||
  368. call->state == AFS_CALL_AWAIT_OP_ID ||
  369. call->state == AFS_CALL_AWAIT_REQUEST ||
  370. call->state == AFS_CALL_AWAIT_ACK) &&
  371. (skb = skb_dequeue(&call->rx_queue))) {
  372. switch (skb->mark) {
  373. case RXRPC_SKB_MARK_DATA:
  374. _debug("Rcv DATA");
  375. last = rxrpc_kernel_is_data_last(skb);
  376. ret = call->type->deliver(call, skb, last);
  377. switch (ret) {
  378. case 0:
  379. if (last &&
  380. call->state == AFS_CALL_AWAIT_REPLY)
  381. call->state = AFS_CALL_COMPLETE;
  382. break;
  383. case -ENOTCONN:
  384. abort_code = RX_CALL_DEAD;
  385. goto do_abort;
  386. case -ENOTSUPP:
  387. abort_code = RX_INVALID_OPERATION;
  388. goto do_abort;
  389. default:
  390. abort_code = RXGEN_CC_UNMARSHAL;
  391. if (call->state != AFS_CALL_AWAIT_REPLY)
  392. abort_code = RXGEN_SS_UNMARSHAL;
  393. do_abort:
  394. rxrpc_kernel_abort_call(call->rxcall,
  395. abort_code);
  396. call->error = ret;
  397. call->state = AFS_CALL_ERROR;
  398. break;
  399. }
  400. afs_data_delivered(skb);
  401. skb = NULL;
  402. continue;
  403. case RXRPC_SKB_MARK_FINAL_ACK:
  404. _debug("Rcv ACK");
  405. call->state = AFS_CALL_COMPLETE;
  406. break;
  407. case RXRPC_SKB_MARK_BUSY:
  408. _debug("Rcv BUSY");
  409. call->error = -EBUSY;
  410. call->state = AFS_CALL_BUSY;
  411. break;
  412. case RXRPC_SKB_MARK_REMOTE_ABORT:
  413. abort_code = rxrpc_kernel_get_abort_code(skb);
  414. call->error = call->type->abort_to_error(abort_code);
  415. call->state = AFS_CALL_ABORTED;
  416. _debug("Rcv ABORT %u -> %d", abort_code, call->error);
  417. break;
  418. case RXRPC_SKB_MARK_NET_ERROR:
  419. call->error = -rxrpc_kernel_get_error_number(skb);
  420. call->state = AFS_CALL_ERROR;
  421. _debug("Rcv NET ERROR %d", call->error);
  422. break;
  423. case RXRPC_SKB_MARK_LOCAL_ERROR:
  424. call->error = -rxrpc_kernel_get_error_number(skb);
  425. call->state = AFS_CALL_ERROR;
  426. _debug("Rcv LOCAL ERROR %d", call->error);
  427. break;
  428. default:
  429. BUG();
  430. break;
  431. }
  432. afs_free_skb(skb);
  433. }
  434. /* make sure the queue is empty if the call is done with (we might have
  435. * aborted the call early because of an unmarshalling error) */
  436. if (call->state >= AFS_CALL_COMPLETE) {
  437. while ((skb = skb_dequeue(&call->rx_queue)))
  438. afs_free_skb(skb);
  439. if (call->incoming) {
  440. rxrpc_kernel_end_call(call->rxcall);
  441. call->rxcall = NULL;
  442. call->type->destructor(call);
  443. afs_free_call(call);
  444. }
  445. }
  446. _leave("");
  447. }
  448. /*
  449. * wait synchronously for a call to complete
  450. */
  451. static int afs_wait_for_call_to_complete(struct afs_call *call)
  452. {
  453. struct sk_buff *skb;
  454. int ret;
  455. DECLARE_WAITQUEUE(myself, current);
  456. _enter("");
  457. add_wait_queue(&call->waitq, &myself);
  458. for (;;) {
  459. set_current_state(TASK_INTERRUPTIBLE);
  460. /* deliver any messages that are in the queue */
  461. if (!skb_queue_empty(&call->rx_queue)) {
  462. __set_current_state(TASK_RUNNING);
  463. afs_deliver_to_call(call);
  464. continue;
  465. }
  466. ret = call->error;
  467. if (call->state >= AFS_CALL_COMPLETE)
  468. break;
  469. ret = -EINTR;
  470. if (signal_pending(current))
  471. break;
  472. schedule();
  473. }
  474. remove_wait_queue(&call->waitq, &myself);
  475. __set_current_state(TASK_RUNNING);
  476. /* kill the call */
  477. if (call->state < AFS_CALL_COMPLETE) {
  478. _debug("call incomplete");
  479. rxrpc_kernel_abort_call(call->rxcall, RX_CALL_DEAD);
  480. while ((skb = skb_dequeue(&call->rx_queue)))
  481. afs_free_skb(skb);
  482. }
  483. _debug("call complete");
  484. rxrpc_kernel_end_call(call->rxcall);
  485. call->rxcall = NULL;
  486. call->type->destructor(call);
  487. afs_free_call(call);
  488. _leave(" = %d", ret);
  489. return ret;
  490. }
  491. /*
  492. * wake up a waiting call
  493. */
  494. static void afs_wake_up_call_waiter(struct afs_call *call)
  495. {
  496. wake_up(&call->waitq);
  497. }
  498. /*
  499. * wake up an asynchronous call
  500. */
  501. static void afs_wake_up_async_call(struct afs_call *call)
  502. {
  503. _enter("");
  504. queue_work(afs_async_calls, &call->async_work);
  505. }
  506. /*
  507. * put a call into asynchronous mode
  508. * - mustn't touch the call descriptor as the call my have completed by the
  509. * time we get here
  510. */
  511. static int afs_dont_wait_for_call_to_complete(struct afs_call *call)
  512. {
  513. _enter("");
  514. return -EINPROGRESS;
  515. }
  516. /*
  517. * delete an asynchronous call
  518. */
  519. static void afs_delete_async_call(struct work_struct *work)
  520. {
  521. struct afs_call *call =
  522. container_of(work, struct afs_call, async_work);
  523. _enter("");
  524. afs_free_call(call);
  525. _leave("");
  526. }
  527. /*
  528. * perform processing on an asynchronous call
  529. * - on a multiple-thread workqueue this work item may try to run on several
  530. * CPUs at the same time
  531. */
  532. static void afs_process_async_call(struct work_struct *work)
  533. {
  534. struct afs_call *call =
  535. container_of(work, struct afs_call, async_work);
  536. _enter("");
  537. if (!skb_queue_empty(&call->rx_queue))
  538. afs_deliver_to_call(call);
  539. if (call->state >= AFS_CALL_COMPLETE && call->wait_mode) {
  540. if (call->wait_mode->async_complete)
  541. call->wait_mode->async_complete(call->reply,
  542. call->error);
  543. call->reply = NULL;
  544. /* kill the call */
  545. rxrpc_kernel_end_call(call->rxcall);
  546. call->rxcall = NULL;
  547. if (call->type->destructor)
  548. call->type->destructor(call);
  549. /* we can't just delete the call because the work item may be
  550. * queued */
  551. PREPARE_WORK(&call->async_work, afs_delete_async_call);
  552. queue_work(afs_async_calls, &call->async_work);
  553. }
  554. _leave("");
  555. }
  556. /*
  557. * empty a socket buffer into a flat reply buffer
  558. */
  559. void afs_transfer_reply(struct afs_call *call, struct sk_buff *skb)
  560. {
  561. size_t len = skb->len;
  562. if (skb_copy_bits(skb, 0, call->buffer + call->reply_size, len) < 0)
  563. BUG();
  564. call->reply_size += len;
  565. }
  566. /*
  567. * accept the backlog of incoming calls
  568. */
  569. static void afs_collect_incoming_call(struct work_struct *work)
  570. {
  571. struct rxrpc_call *rxcall;
  572. struct afs_call *call = NULL;
  573. struct sk_buff *skb;
  574. while ((skb = skb_dequeue(&afs_incoming_calls))) {
  575. _debug("new call");
  576. /* don't need the notification */
  577. afs_free_skb(skb);
  578. if (!call) {
  579. call = kzalloc(sizeof(struct afs_call), GFP_KERNEL);
  580. if (!call) {
  581. rxrpc_kernel_reject_call(afs_socket);
  582. return;
  583. }
  584. INIT_WORK(&call->async_work, afs_process_async_call);
  585. call->wait_mode = &afs_async_incoming_call;
  586. call->type = &afs_RXCMxxxx;
  587. init_waitqueue_head(&call->waitq);
  588. skb_queue_head_init(&call->rx_queue);
  589. call->state = AFS_CALL_AWAIT_OP_ID;
  590. _debug("CALL %p{%s} [%d]",
  591. call, call->type->name,
  592. atomic_read(&afs_outstanding_calls));
  593. atomic_inc(&afs_outstanding_calls);
  594. }
  595. rxcall = rxrpc_kernel_accept_call(afs_socket,
  596. (unsigned long) call);
  597. if (!IS_ERR(rxcall)) {
  598. call->rxcall = rxcall;
  599. call = NULL;
  600. }
  601. }
  602. if (call)
  603. afs_free_call(call);
  604. }
  605. /*
  606. * grab the operation ID from an incoming cache manager call
  607. */
  608. static int afs_deliver_cm_op_id(struct afs_call *call, struct sk_buff *skb,
  609. bool last)
  610. {
  611. size_t len = skb->len;
  612. void *oibuf = (void *) &call->operation_ID;
  613. _enter("{%u},{%zu},%d", call->offset, len, last);
  614. ASSERTCMP(call->offset, <, 4);
  615. /* the operation ID forms the first four bytes of the request data */
  616. len = min_t(size_t, len, 4 - call->offset);
  617. if (skb_copy_bits(skb, 0, oibuf + call->offset, len) < 0)
  618. BUG();
  619. if (!pskb_pull(skb, len))
  620. BUG();
  621. call->offset += len;
  622. if (call->offset < 4) {
  623. if (last) {
  624. _leave(" = -EBADMSG [op ID short]");
  625. return -EBADMSG;
  626. }
  627. _leave(" = 0 [incomplete]");
  628. return 0;
  629. }
  630. call->state = AFS_CALL_AWAIT_REQUEST;
  631. /* ask the cache manager to route the call (it'll change the call type
  632. * if successful) */
  633. if (!afs_cm_incoming_call(call))
  634. return -ENOTSUPP;
  635. /* pass responsibility for the remainer of this message off to the
  636. * cache manager op */
  637. return call->type->deliver(call, skb, last);
  638. }
  639. /*
  640. * send an empty reply
  641. */
  642. void afs_send_empty_reply(struct afs_call *call)
  643. {
  644. struct msghdr msg;
  645. struct iovec iov[1];
  646. _enter("");
  647. iov[0].iov_base = NULL;
  648. iov[0].iov_len = 0;
  649. msg.msg_name = NULL;
  650. msg.msg_namelen = 0;
  651. msg.msg_iov = iov;
  652. msg.msg_iovlen = 0;
  653. msg.msg_control = NULL;
  654. msg.msg_controllen = 0;
  655. msg.msg_flags = 0;
  656. call->state = AFS_CALL_AWAIT_ACK;
  657. switch (rxrpc_kernel_send_data(call->rxcall, &msg, 0)) {
  658. case 0:
  659. _leave(" [replied]");
  660. return;
  661. case -ENOMEM:
  662. _debug("oom");
  663. rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
  664. default:
  665. rxrpc_kernel_end_call(call->rxcall);
  666. call->rxcall = NULL;
  667. call->type->destructor(call);
  668. afs_free_call(call);
  669. _leave(" [error]");
  670. return;
  671. }
  672. }
  673. /*
  674. * send a simple reply
  675. */
  676. void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len)
  677. {
  678. struct msghdr msg;
  679. struct iovec iov[1];
  680. int n;
  681. _enter("");
  682. iov[0].iov_base = (void *) buf;
  683. iov[0].iov_len = len;
  684. msg.msg_name = NULL;
  685. msg.msg_namelen = 0;
  686. msg.msg_iov = iov;
  687. msg.msg_iovlen = 1;
  688. msg.msg_control = NULL;
  689. msg.msg_controllen = 0;
  690. msg.msg_flags = 0;
  691. call->state = AFS_CALL_AWAIT_ACK;
  692. n = rxrpc_kernel_send_data(call->rxcall, &msg, len);
  693. if (n >= 0) {
  694. _leave(" [replied]");
  695. return;
  696. }
  697. if (n == -ENOMEM) {
  698. _debug("oom");
  699. rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
  700. }
  701. rxrpc_kernel_end_call(call->rxcall);
  702. call->rxcall = NULL;
  703. call->type->destructor(call);
  704. afs_free_call(call);
  705. _leave(" [error]");
  706. }
  707. /*
  708. * extract a piece of data from the received data socket buffers
  709. */
  710. int afs_extract_data(struct afs_call *call, struct sk_buff *skb,
  711. bool last, void *buf, size_t count)
  712. {
  713. size_t len = skb->len;
  714. _enter("{%u},{%zu},%d,,%zu", call->offset, len, last, count);
  715. ASSERTCMP(call->offset, <, count);
  716. len = min_t(size_t, len, count - call->offset);
  717. if (skb_copy_bits(skb, 0, buf + call->offset, len) < 0 ||
  718. !pskb_pull(skb, len))
  719. BUG();
  720. call->offset += len;
  721. if (call->offset < count) {
  722. if (last) {
  723. _leave(" = -EBADMSG [%d < %zu]", call->offset, count);
  724. return -EBADMSG;
  725. }
  726. _leave(" = -EAGAIN");
  727. return -EAGAIN;
  728. }
  729. return 0;
  730. }