stream.c 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287
  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. while (1) {
  50. if (sk->sk_err)
  51. return sock_error(sk);
  52. if ((1 << sk->sk_state) & ~(TCPF_SYN_SENT | TCPF_SYN_RECV))
  53. return -EPIPE;
  54. if (!*timeo_p)
  55. return -EAGAIN;
  56. if (signal_pending(tsk))
  57. return sock_intr_errno(*timeo_p);
  58. prepare_to_wait(sk->sk_sleep, &wait, TASK_INTERRUPTIBLE);
  59. sk->sk_write_pending++;
  60. if (sk_wait_event(sk, timeo_p,
  61. !((1 << sk->sk_state) &
  62. ~(TCPF_ESTABLISHED | TCPF_CLOSE_WAIT))))
  63. break;
  64. finish_wait(sk->sk_sleep, &wait);
  65. sk->sk_write_pending--;
  66. }
  67. return 0;
  68. }
  69. EXPORT_SYMBOL(sk_stream_wait_connect);
  70. /**
  71. * sk_stream_closing - Return 1 if we still have things to send in our buffers.
  72. * @sk: socket to verify
  73. */
  74. static inline int sk_stream_closing(struct sock *sk)
  75. {
  76. return (1 << sk->sk_state) &
  77. (TCPF_FIN_WAIT1 | TCPF_CLOSING | TCPF_LAST_ACK);
  78. }
  79. void sk_stream_wait_close(struct sock *sk, long timeout)
  80. {
  81. if (timeout) {
  82. DEFINE_WAIT(wait);
  83. do {
  84. prepare_to_wait(sk->sk_sleep, &wait,
  85. TASK_INTERRUPTIBLE);
  86. if (sk_wait_event(sk, &timeout, !sk_stream_closing(sk)))
  87. break;
  88. } while (!signal_pending(current) && timeout);
  89. finish_wait(sk->sk_sleep, &wait);
  90. }
  91. }
  92. EXPORT_SYMBOL(sk_stream_wait_close);
  93. /**
  94. * sk_stream_wait_memory - Wait for more memory for a socket
  95. * @sk: socket to wait for memory
  96. * @timeo_p: for how long
  97. */
  98. int sk_stream_wait_memory(struct sock *sk, long *timeo_p)
  99. {
  100. int err = 0;
  101. long vm_wait = 0;
  102. long current_timeo = *timeo_p;
  103. DEFINE_WAIT(wait);
  104. if (sk_stream_memory_free(sk))
  105. current_timeo = vm_wait = (net_random() % (HZ / 5)) + 2;
  106. while (1) {
  107. set_bit(SOCK_ASYNC_NOSPACE, &sk->sk_socket->flags);
  108. prepare_to_wait(sk->sk_sleep, &wait, TASK_INTERRUPTIBLE);
  109. if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
  110. goto do_error;
  111. if (!*timeo_p)
  112. goto do_nonblock;
  113. if (signal_pending(current))
  114. goto do_interrupted;
  115. clear_bit(SOCK_ASYNC_NOSPACE, &sk->sk_socket->flags);
  116. if (sk_stream_memory_free(sk) && !vm_wait)
  117. break;
  118. set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
  119. sk->sk_write_pending++;
  120. sk_wait_event(sk, &current_timeo, sk_stream_memory_free(sk) &&
  121. vm_wait);
  122. sk->sk_write_pending--;
  123. if (vm_wait) {
  124. vm_wait -= current_timeo;
  125. current_timeo = *timeo_p;
  126. if (current_timeo != MAX_SCHEDULE_TIMEOUT &&
  127. (current_timeo -= vm_wait) < 0)
  128. current_timeo = 0;
  129. vm_wait = 0;
  130. }
  131. *timeo_p = current_timeo;
  132. }
  133. out:
  134. finish_wait(sk->sk_sleep, &wait);
  135. return err;
  136. do_error:
  137. err = -EPIPE;
  138. goto out;
  139. do_nonblock:
  140. err = -EAGAIN;
  141. goto out;
  142. do_interrupted:
  143. err = sock_intr_errno(*timeo_p);
  144. goto out;
  145. }
  146. EXPORT_SYMBOL(sk_stream_wait_memory);
  147. void sk_stream_rfree(struct sk_buff *skb)
  148. {
  149. struct sock *sk = skb->sk;
  150. atomic_sub(skb->truesize, &sk->sk_rmem_alloc);
  151. sk->sk_forward_alloc += skb->truesize;
  152. }
  153. EXPORT_SYMBOL(sk_stream_rfree);
  154. int sk_stream_error(struct sock *sk, int flags, int err)
  155. {
  156. if (err == -EPIPE)
  157. err = sock_error(sk) ? : -EPIPE;
  158. if (err == -EPIPE && !(flags & MSG_NOSIGNAL))
  159. send_sig(SIGPIPE, current, 0);
  160. return err;
  161. }
  162. EXPORT_SYMBOL(sk_stream_error);
  163. void __sk_stream_mem_reclaim(struct sock *sk)
  164. {
  165. if (sk->sk_forward_alloc >= SK_STREAM_MEM_QUANTUM) {
  166. atomic_sub(sk->sk_forward_alloc / SK_STREAM_MEM_QUANTUM,
  167. sk->sk_prot->memory_allocated);
  168. sk->sk_forward_alloc &= SK_STREAM_MEM_QUANTUM - 1;
  169. if (*sk->sk_prot->memory_pressure &&
  170. (atomic_read(sk->sk_prot->memory_allocated) <
  171. sk->sk_prot->sysctl_mem[0]))
  172. *sk->sk_prot->memory_pressure = 0;
  173. }
  174. }
  175. EXPORT_SYMBOL(__sk_stream_mem_reclaim);
  176. int sk_stream_mem_schedule(struct sock *sk, int size, int kind)
  177. {
  178. int amt = sk_stream_pages(size);
  179. sk->sk_forward_alloc += amt * SK_STREAM_MEM_QUANTUM;
  180. atomic_add(amt, sk->sk_prot->memory_allocated);
  181. /* Under limit. */
  182. if (atomic_read(sk->sk_prot->memory_allocated) < sk->sk_prot->sysctl_mem[0]) {
  183. if (*sk->sk_prot->memory_pressure)
  184. *sk->sk_prot->memory_pressure = 0;
  185. return 1;
  186. }
  187. /* Over hard limit. */
  188. if (atomic_read(sk->sk_prot->memory_allocated) > sk->sk_prot->sysctl_mem[2]) {
  189. sk->sk_prot->enter_memory_pressure();
  190. goto suppress_allocation;
  191. }
  192. /* Under pressure. */
  193. if (atomic_read(sk->sk_prot->memory_allocated) > sk->sk_prot->sysctl_mem[1])
  194. sk->sk_prot->enter_memory_pressure();
  195. if (kind) {
  196. if (atomic_read(&sk->sk_rmem_alloc) < sk->sk_prot->sysctl_rmem[0])
  197. return 1;
  198. } else if (sk->sk_wmem_queued < sk->sk_prot->sysctl_wmem[0])
  199. return 1;
  200. if (!*sk->sk_prot->memory_pressure ||
  201. sk->sk_prot->sysctl_mem[2] > atomic_read(sk->sk_prot->sockets_allocated) *
  202. sk_stream_pages(sk->sk_wmem_queued +
  203. atomic_read(&sk->sk_rmem_alloc) +
  204. sk->sk_forward_alloc))
  205. return 1;
  206. suppress_allocation:
  207. if (!kind) {
  208. sk_stream_moderate_sndbuf(sk);
  209. /* Fail only if socket is _under_ its sndbuf.
  210. * In this case we cannot block, so that we have to fail.
  211. */
  212. if (sk->sk_wmem_queued + size >= sk->sk_sndbuf)
  213. return 1;
  214. }
  215. /* Alas. Undo changes. */
  216. sk->sk_forward_alloc -= amt * SK_STREAM_MEM_QUANTUM;
  217. atomic_sub(amt, sk->sk_prot->memory_allocated);
  218. return 0;
  219. }
  220. EXPORT_SYMBOL(sk_stream_mem_schedule);
  221. void sk_stream_kill_queues(struct sock *sk)
  222. {
  223. /* First the read buffer. */
  224. __skb_queue_purge(&sk->sk_receive_queue);
  225. /* Next, the error queue. */
  226. __skb_queue_purge(&sk->sk_error_queue);
  227. /* Next, the write queue. */
  228. BUG_TRAP(skb_queue_empty(&sk->sk_write_queue));
  229. /* Account for returned memory. */
  230. sk_stream_mem_reclaim(sk);
  231. BUG_TRAP(!sk->sk_wmem_queued);
  232. BUG_TRAP(!sk->sk_forward_alloc);
  233. /* It is _impossible_ for the backlog to contain anything
  234. * when we get here. All user references to this socket
  235. * have gone away, only the net layer knows can touch it.
  236. */
  237. }
  238. EXPORT_SYMBOL(sk_stream_kill_queues);