xref: /unit/src/nxt_conn_read.c (revision 629:b3e55c647741)
1 
2 /*
3  * Copyright (C) Igor Sysoev
4  * Copyright (C) NGINX, Inc.
5  */
6 
7 #include <nxt_main.h>
8 
9 
10 void
11 nxt_conn_wait(nxt_conn_t *c)
12 {
13     nxt_event_engine_t      *engine;
14     const nxt_conn_state_t  *state;
15 
16     nxt_debug(c->socket.task, "conn wait fd:%d rdy:%d",
17               c->socket.fd, c->socket.read_ready);
18 
19     engine = c->socket.task->thread->engine;
20     state = c->read_state;
21 
22     if (c->socket.read_ready) {
23         nxt_work_queue_add(&engine->fast_work_queue, state->ready_handler,
24                            c->socket.task, c, c->socket.data);
25         return;
26     }
27 
28     c->socket.read_handler = state->ready_handler;
29     c->socket.error_handler = state->error_handler;
30 
31     nxt_conn_timer(engine, c, state, &c->read_timer);
32 
33     nxt_fd_event_enable_read(engine, &c->socket);
34 }
35 
36 
37 void
38 nxt_conn_io_read(nxt_task_t *task, void *obj, void *data)
39 {
40     ssize_t                 n;
41     nxt_conn_t              *c;
42     nxt_work_queue_t        *wq;
43     nxt_event_engine_t      *engine;
44     nxt_work_handler_t      handler;
45     const nxt_conn_state_t  *state;
46 
47     c = obj;
48 
49     nxt_debug(task, "conn read fd:%d rdy:%d cl:%d",
50               c->socket.fd, c->socket.read_ready, c->socket.closed);
51 
52     engine = task->thread->engine;
53 
54     state = c->read_state;
55 
56     if (c->socket.read_ready) {
57 
58         if (state->io_read_handler == NULL) {
59             n = c->io->recvbuf(c, c->read);
60 
61         } else {
62             n = state->io_read_handler(c);
63         }
64 
65         if (n > 0) {
66             c->nbytes = n;
67 
68             nxt_recvbuf_update(c->read, n);
69 
70             nxt_fd_event_block_read(engine, &c->socket);
71 
72             if (state->timer_autoreset) {
73                 nxt_timer_disable(engine, &c->read_timer);
74             }
75 
76             wq = c->read_work_queue;
77             handler = state->ready_handler;
78 
79             nxt_work_queue_add(wq, handler, task, c, data);
80 
81             return;
82         }
83 
84         if (n != NXT_AGAIN) {
85             nxt_fd_event_block_read(engine, &c->socket);
86             nxt_timer_disable(engine, &c->read_timer);
87 
88             wq = &engine->fast_work_queue;
89 
90             handler = (n == 0) ? state->close_handler : state->error_handler;
91 
92             nxt_work_queue_add(wq, handler, task, c, data);
93 
94             return;
95         }
96     }
97 
98     /*
99      * Here c->io->read() is assigned instead of direct nxt_conn_io_read()
100      * because the function can be called by nxt_kqueue_conn_io_read().
101      */
102     c->socket.read_handler = c->io->read;
103     c->socket.error_handler = state->error_handler;
104 
105     if (c->read_timer.state == NXT_TIMER_DISABLED
106         || nxt_fd_event_is_disabled(c->socket.read))
107     {
108         /* Timer may be set or reset. */
109         nxt_conn_timer(engine, c, state, &c->read_timer);
110 
111         if (nxt_fd_event_is_disabled(c->socket.read)) {
112             nxt_fd_event_enable_read(engine, &c->socket);
113         }
114     }
115 
116     return;
117 }
118 
119 
120 ssize_t
121 nxt_conn_io_recvbuf(nxt_conn_t *c, nxt_buf_t *b)
122 {
123     ssize_t                 n;
124     nxt_err_t               err;
125     nxt_uint_t              niov;
126     struct iovec            iov[NXT_IOBUF_MAX];
127     nxt_recvbuf_coalesce_t  rb;
128 
129     rb.buf = b;
130     rb.iobuf = iov;
131     rb.nmax = NXT_IOBUF_MAX;
132     rb.size = 0;
133 
134     niov = nxt_recvbuf_mem_coalesce(&rb);
135 
136     if (niov == 1) {
137         /* Disposal of surplus kernel iovec copy-in operation. */
138         return nxt_conn_io_recv(c, iov->iov_base, iov->iov_len, 0);
139     }
140 
141     for ( ;; ) {
142         n = readv(c->socket.fd, iov, niov);
143 
144         err = (n == -1) ? nxt_socket_errno : 0;
145 
146         nxt_debug(c->socket.task, "readv(%d, %ui): %z", c->socket.fd, niov, n);
147 
148         if (n > 0) {
149             if ((size_t) n < rb.size) {
150                 c->socket.read_ready = 0;
151             }
152 
153             return n;
154         }
155 
156         if (n == 0) {
157             c->socket.closed = 1;
158             c->socket.read_ready = 0;
159             return n;
160         }
161 
162         /* n == -1 */
163 
164         switch (err) {
165 
166         case NXT_EAGAIN:
167             nxt_debug(c->socket.task, "readv() %E", err);
168             c->socket.read_ready = 0;
169             return NXT_AGAIN;
170 
171         case NXT_EINTR:
172             nxt_debug(c->socket.task, "readv() %E", err);
173             continue;
174 
175         default:
176             c->socket.error = err;
177             nxt_log(c->socket.task, nxt_socket_error_level(err),
178                     "readv(%d, %ui) failed %E", c->socket.fd, niov, err);
179 
180             return NXT_ERROR;
181         }
182     }
183 }
184 
185 
186 ssize_t
187 nxt_conn_io_recv(nxt_conn_t *c, void *buf, size_t size, nxt_uint_t flags)
188 {
189     ssize_t    n;
190     nxt_err_t  err;
191 
192     for ( ;; ) {
193         n = recv(c->socket.fd, buf, size, flags);
194 
195         err = (n == -1) ? nxt_socket_errno : 0;
196 
197         nxt_debug(c->socket.task, "recv(%d, %p, %uz, 0x%ui): %z",
198                   c->socket.fd, buf, size, flags, n);
199 
200         if (n > 0) {
201             if ((size_t) n < size) {
202                 c->socket.read_ready = 0;
203             }
204 
205             return n;
206         }
207 
208         if (n == 0) {
209             c->socket.closed = 1;
210             c->socket.read_ready = 0;
211 
212             return n;
213         }
214 
215         /* n == -1 */
216 
217         switch (err) {
218 
219         case NXT_EAGAIN:
220             nxt_debug(c->socket.task, "recv() %E", err);
221             c->socket.read_ready = 0;
222 
223             return NXT_AGAIN;
224 
225         case NXT_EINTR:
226             nxt_debug(c->socket.task, "recv() %E", err);
227             continue;
228 
229         default:
230             c->socket.error = err;
231             nxt_log(c->socket.task, nxt_socket_error_level(err),
232                     "recv(%d, %p, %uz, %ui) failed %E",
233                     c->socket.fd, buf, size, flags, err);
234 
235             return NXT_ERROR;
236         }
237     }
238 }
239