stream.c 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290
  1. /*
  2. * SUCS NET3:
  3. *
  4. * Generic stream handling routines. These are generic for most
  5. * protocols. Even IP. Tonight 8-).
  6. * This is used because TCP, LLC (others too) layer all have mostly
  7. * identical sendmsg() and recvmsg() code.
  8. * So we (will) share it here.
  9. *
  10. * Authors: Arnaldo Carvalho de Melo <acme@conectiva.com.br>
  11. * (from old tcp.c code)
  12. * Alan Cox <alan@redhat.com> (Borrowed comments 8-))
  13. */
  14. #include <linux/module.h>
  15. #include <linux/net.h>
  16. #include <linux/signal.h>
  17. #include <linux/tcp.h>
  18. #include <linux/wait.h>
  19. #include <net/sock.h>
  20. /**
  21. * sk_stream_write_space - stream socket write_space callback.
  22. * @sk: socket
  23. *
  24. * FIXME: write proper description
  25. */
  26. void sk_stream_write_space(struct sock *sk)
  27. {
  28. struct socket *sock = sk->sk_socket;
  29. if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) && sock) {
  30. clear_bit(SOCK_NOSPACE, &sock->flags);
  31. if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
  32. wake_up_interruptible(sk->sk_sleep);
  33. if (sock->fasync_list && !(sk->sk_shutdown & SEND_SHUTDOWN))
  34. sock_wake_async(sock, 2, POLL_OUT);
  35. }
  36. }
  37. EXPORT_SYMBOL(sk_stream_write_space);
  38. /**
  39. * sk_stream_wait_connect - Wait for a socket to get into the connected state
  40. * @sk: sock to wait on
  41. * @timeo_p: for how long to wait
  42. *
  43. * Must be called with the socket locked.
  44. */
  45. int sk_stream_wait_connect(struct sock *sk, long *timeo_p)
  46. {
  47. struct task_struct *tsk = current;
  48. DEFINE_WAIT(wait);
  49. int done;
  50. do {
  51. int err = sock_error(sk);
  52. if (err)
  53. return err;
  54. if ((1 << sk->sk_state) & ~(TCPF_SYN_SENT | TCPF_SYN_RECV))
  55. return -EPIPE;
  56. if (!*timeo_p)
  57. return -EAGAIN;
  58. if (signal_pending(tsk))
  59. return sock_intr_errno(*timeo_p);
  60. prepare_to_wait(sk->sk_sleep, &wait, TASK_INTERRUPTIBLE);
  61. sk->sk_write_pending++;
  62. done = sk_wait_event(sk, timeo_p,
  63. !sk->sk_err &&
  64. !((1 << sk->sk_state) &
  65. ~(TCPF_ESTABLISHED | TCPF_CLOSE_WAIT)));
  66. finish_wait(sk->sk_sleep, &wait);
  67. sk->sk_write_pending--;
  68. } while (!done);
  69. return 0;
  70. }
  71. EXPORT_SYMBOL(sk_stream_wait_connect);
  72. /**
  73. * sk_stream_closing - Return 1 if we still have things to send in our buffers.
  74. * @sk: socket to verify
  75. */
  76. static inline int sk_stream_closing(struct sock *sk)
  77. {
  78. return (1 << sk->sk_state) &
  79. (TCPF_FIN_WAIT1 | TCPF_CLOSING | TCPF_LAST_ACK);
  80. }
  81. void sk_stream_wait_close(struct sock *sk, long timeout)
  82. {
  83. if (timeout) {
  84. DEFINE_WAIT(wait);
  85. do {
  86. prepare_to_wait(sk->sk_sleep, &wait,
  87. TASK_INTERRUPTIBLE);
  88. if (sk_wait_event(sk, &timeout, !sk_stream_closing(sk)))
  89. break;
  90. } while (!signal_pending(current) && timeout);
  91. finish_wait(sk->sk_sleep, &wait);
  92. }
  93. }
  94. EXPORT_SYMBOL(sk_stream_wait_close);
  95. /**
  96. * sk_stream_wait_memory - Wait for more memory for a socket
  97. * @sk: socket to wait for memory
  98. * @timeo_p: for how long
  99. */
  100. int sk_stream_wait_memory(struct sock *sk, long *timeo_p)
  101. {
  102. int err = 0;
  103. long vm_wait = 0;
  104. long current_timeo = *timeo_p;
  105. DEFINE_WAIT(wait);
  106. if (sk_stream_memory_free(sk))
  107. current_timeo = vm_wait = (net_random() % (HZ / 5)) + 2;
  108. while (1) {
  109. set_bit(SOCK_ASYNC_NOSPACE, &sk->sk_socket->flags);
  110. prepare_to_wait(sk->sk_sleep, &wait, TASK_INTERRUPTIBLE);
  111. if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
  112. goto do_error;
  113. if (!*timeo_p)
  114. goto do_nonblock;
  115. if (signal_pending(current))
  116. goto do_interrupted;
  117. clear_bit(SOCK_ASYNC_NOSPACE, &sk->sk_socket->flags);
  118. if (sk_stream_memory_free(sk) && !vm_wait)
  119. break;
  120. set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
  121. sk->sk_write_pending++;
  122. sk_wait_event(sk, &current_timeo, !sk->sk_err &&
  123. !(sk->sk_shutdown & SEND_SHUTDOWN) &&
  124. sk_stream_memory_free(sk) &&
  125. vm_wait);
  126. sk->sk_write_pending--;
  127. if (vm_wait) {
  128. vm_wait -= current_timeo;
  129. current_timeo = *timeo_p;
  130. if (current_timeo != MAX_SCHEDULE_TIMEOUT &&
  131. (current_timeo -= vm_wait) < 0)
  132. current_timeo = 0;
  133. vm_wait = 0;
  134. }
  135. *timeo_p = current_timeo;
  136. }
  137. out:
  138. finish_wait(sk->sk_sleep, &wait);
  139. return err;
  140. do_error:
  141. err = -EPIPE;
  142. goto out;
  143. do_nonblock:
  144. err = -EAGAIN;
  145. goto out;
  146. do_interrupted:
  147. err = sock_intr_errno(*timeo_p);
  148. goto out;
  149. }
  150. EXPORT_SYMBOL(sk_stream_wait_memory);
  151. void sk_stream_rfree(struct sk_buff *skb)
  152. {
  153. struct sock *sk = skb->sk;
  154. skb_truesize_check(skb);
  155. atomic_sub(skb->truesize, &sk->sk_rmem_alloc);
  156. sk->sk_forward_alloc += skb->truesize;
  157. }
  158. EXPORT_SYMBOL(sk_stream_rfree);
  159. int sk_stream_error(struct sock *sk, int flags, int err)
  160. {
  161. if (err == -EPIPE)
  162. err = sock_error(sk) ? : -EPIPE;
  163. if (err == -EPIPE && !(flags & MSG_NOSIGNAL))
  164. send_sig(SIGPIPE, current, 0);
  165. return err;
  166. }
  167. EXPORT_SYMBOL(sk_stream_error);
  168. void __sk_stream_mem_reclaim(struct sock *sk)
  169. {
  170. atomic_sub(sk->sk_forward_alloc / SK_STREAM_MEM_QUANTUM,
  171. sk->sk_prot->memory_allocated);
  172. sk->sk_forward_alloc &= SK_STREAM_MEM_QUANTUM - 1;
  173. if (*sk->sk_prot->memory_pressure &&
  174. (atomic_read(sk->sk_prot->memory_allocated) <
  175. sk->sk_prot->sysctl_mem[0]))
  176. *sk->sk_prot->memory_pressure = 0;
  177. }
  178. EXPORT_SYMBOL(__sk_stream_mem_reclaim);
  179. int sk_stream_mem_schedule(struct sock *sk, int size, int kind)
  180. {
  181. int amt = sk_stream_pages(size);
  182. sk->sk_forward_alloc += amt * SK_STREAM_MEM_QUANTUM;
  183. atomic_add(amt, sk->sk_prot->memory_allocated);
  184. /* Under limit. */
  185. if (atomic_read(sk->sk_prot->memory_allocated) < sk->sk_prot->sysctl_mem[0]) {
  186. if (*sk->sk_prot->memory_pressure)
  187. *sk->sk_prot->memory_pressure = 0;
  188. return 1;
  189. }
  190. /* Over hard limit. */
  191. if (atomic_read(sk->sk_prot->memory_allocated) > sk->sk_prot->sysctl_mem[2]) {
  192. sk->sk_prot->enter_memory_pressure();
  193. goto suppress_allocation;
  194. }
  195. /* Under pressure. */
  196. if (atomic_read(sk->sk_prot->memory_allocated) > sk->sk_prot->sysctl_mem[1])
  197. sk->sk_prot->enter_memory_pressure();
  198. if (kind) {
  199. if (atomic_read(&sk->sk_rmem_alloc) < sk->sk_prot->sysctl_rmem[0])
  200. return 1;
  201. } else if (sk->sk_wmem_queued < sk->sk_prot->sysctl_wmem[0])
  202. return 1;
  203. if (!*sk->sk_prot->memory_pressure ||
  204. sk->sk_prot->sysctl_mem[2] > atomic_read(sk->sk_prot->sockets_allocated) *
  205. sk_stream_pages(sk->sk_wmem_queued +
  206. atomic_read(&sk->sk_rmem_alloc) +
  207. sk->sk_forward_alloc))
  208. return 1;
  209. suppress_allocation:
  210. if (!kind) {
  211. sk_stream_moderate_sndbuf(sk);
  212. /* Fail only if socket is _under_ its sndbuf.
  213. * In this case we cannot block, so that we have to fail.
  214. */
  215. if (sk->sk_wmem_queued + size >= sk->sk_sndbuf)
  216. return 1;
  217. }
  218. /* Alas. Undo changes. */
  219. sk->sk_forward_alloc -= amt * SK_STREAM_MEM_QUANTUM;
  220. atomic_sub(amt, sk->sk_prot->memory_allocated);
  221. return 0;
  222. }
  223. EXPORT_SYMBOL(sk_stream_mem_schedule);
  224. void sk_stream_kill_queues(struct sock *sk)
  225. {
  226. /* First the read buffer. */
  227. __skb_queue_purge(&sk->sk_receive_queue);
  228. /* Next, the error queue. */
  229. __skb_queue_purge(&sk->sk_error_queue);
  230. /* Next, the write queue. */
  231. BUG_TRAP(skb_queue_empty(&sk->sk_write_queue));
  232. /* Account for returned memory. */
  233. sk_stream_mem_reclaim(sk);
  234. BUG_TRAP(!sk->sk_wmem_queued);
  235. BUG_TRAP(!sk->sk_forward_alloc);
  236. /* It is _impossible_ for the backlog to contain anything
  237. * when we get here. All user references to this socket
  238. * have gone away, only the net layer knows can touch it.
  239. */
  240. }
  241. EXPORT_SYMBOL(sk_stream_kill_queues);