rxrpc.c 20 KB

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