netdev: Add 'patch' userspace implementation.
[openvswitch] / lib / vconn-stream.c
1 /*
2  * Copyright (c) 2008, 2009, 2010 Nicira Networks.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at:
7  *
8  *     http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16
17 #include <config.h>
18 #include <assert.h>
19 #include <errno.h>
20 #include <poll.h>
21 #include <stdlib.h>
22 #include <string.h>
23 #include <sys/types.h>
24 #include <unistd.h>
25 #include "fatal-signal.h"
26 #include "leak-checker.h"
27 #include "ofpbuf.h"
28 #include "openflow/openflow.h"
29 #include "poll-loop.h"
30 #include "socket-util.h"
31 #include "stream.h"
32 #include "util.h"
33 #include "vconn-provider.h"
34 #include "vconn.h"
35
36 #include "vlog.h"
37 #define THIS_MODULE VLM_vconn_stream
38
39 /* Active stream socket vconn. */
40
41 struct vconn_stream
42 {
43     struct vconn vconn;
44     struct stream *stream;
45     struct ofpbuf *rxbuf;
46     struct ofpbuf *txbuf;
47     int n_packets;
48 };
49
50 static struct vconn_class stream_vconn_class;
51
52 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 25);
53
54 static void vconn_stream_clear_txbuf(struct vconn_stream *);
55
56 static struct vconn *
57 vconn_stream_new(struct stream *stream, int connect_status)
58 {
59     struct vconn_stream *s;
60
61     s = xmalloc(sizeof *s);
62     vconn_init(&s->vconn, &stream_vconn_class, connect_status,
63                stream_get_name(stream));
64     s->stream = stream;
65     s->txbuf = NULL;
66     s->rxbuf = NULL;
67     s->n_packets = 0;
68     s->vconn.remote_ip = stream_get_remote_ip(stream);
69     s->vconn.remote_port = stream_get_remote_port(stream);
70     s->vconn.local_ip = stream_get_local_ip(stream);
71     s->vconn.local_port = stream_get_local_port(stream);
72     return &s->vconn;
73 }
74
75 /* Creates a new vconn that will send and receive data on a stream named 'name'
76  * and stores a pointer to the vconn in '*vconnp'.
77  *
78  * Returns 0 if successful, otherwise a positive errno value. */
79 static int
80 vconn_stream_open(const char *name, char *suffix OVS_UNUSED,
81                   struct vconn **vconnp)
82 {
83     struct stream *stream;
84     int error;
85
86     error = stream_open_with_default_ports(name, OFP_TCP_PORT, OFP_SSL_PORT,
87                                            &stream);
88
89     if (error && error != EAGAIN) {
90         return error;
91     }
92
93     *vconnp = vconn_stream_new(stream, error);
94     return 0;
95 }
96
97 static struct vconn_stream *
98 vconn_stream_cast(struct vconn *vconn)
99 {
100     return CONTAINER_OF(vconn, struct vconn_stream, vconn);
101 }
102
103 static void
104 vconn_stream_close(struct vconn *vconn)
105 {
106     struct vconn_stream *s = vconn_stream_cast(vconn);
107
108     if ((vconn->error == EPROTO || s->n_packets < 1) && s->rxbuf) {
109         stream_report_content(s->rxbuf->data, s->rxbuf->size, STREAM_OPENFLOW,
110                               THIS_MODULE, vconn_get_name(vconn));
111     }
112
113     stream_close(s->stream);
114     vconn_stream_clear_txbuf(s);
115     ofpbuf_delete(s->rxbuf);
116     free(s);
117 }
118
119 static int
120 vconn_stream_connect(struct vconn *vconn)
121 {
122     struct vconn_stream *s = vconn_stream_cast(vconn);
123     return stream_connect(s->stream);
124 }
125
126 static int
127 vconn_stream_recv__(struct vconn_stream *s, int rx_len)
128 {
129     struct ofpbuf *rx = s->rxbuf;
130     int want_bytes, retval;
131
132     want_bytes = rx_len - rx->size;
133     ofpbuf_prealloc_tailroom(rx, want_bytes);
134     retval = stream_recv(s->stream, ofpbuf_tail(rx), want_bytes);
135     if (retval > 0) {
136         rx->size += retval;
137         return retval == want_bytes ? 0 : EAGAIN;
138     } else if (retval == 0) {
139         if (rx->size) {
140             VLOG_ERR_RL(&rl, "connection dropped mid-packet");
141             return EPROTO;
142         }
143         return EOF;
144     } else {
145         return -retval;
146     }
147 }
148
149 static int
150 vconn_stream_recv(struct vconn *vconn, struct ofpbuf **bufferp)
151 {
152     struct vconn_stream *s = vconn_stream_cast(vconn);
153     const struct ofp_header *oh;
154     int rx_len;
155
156     /* Allocate new receive buffer if we don't have one. */
157     if (s->rxbuf == NULL) {
158         s->rxbuf = ofpbuf_new(1564);
159     }
160
161     /* Read ofp_header. */
162     if (s->rxbuf->size < sizeof(struct ofp_header)) {
163         int retval = vconn_stream_recv__(s, sizeof(struct ofp_header));
164         if (retval) {
165             return retval;
166         }
167     }
168
169     /* Read payload. */
170     oh = s->rxbuf->data;
171     rx_len = ntohs(oh->length);
172     if (rx_len < sizeof(struct ofp_header)) {
173         VLOG_ERR_RL(&rl, "received too-short ofp_header (%zu bytes)",
174                     rx_len);
175         return EPROTO;
176     } else if (s->rxbuf->size < rx_len) {
177         int retval = vconn_stream_recv__(s, rx_len);
178         if (retval) {
179             return retval;
180         }
181     }
182
183     s->n_packets++;
184     *bufferp = s->rxbuf;
185     s->rxbuf = NULL;
186     return 0;
187 }
188
189 static void
190 vconn_stream_clear_txbuf(struct vconn_stream *s)
191 {
192     ofpbuf_delete(s->txbuf);
193     s->txbuf = NULL;
194 }
195
196 static int
197 vconn_stream_send(struct vconn *vconn, struct ofpbuf *buffer)
198 {
199     struct vconn_stream *s = vconn_stream_cast(vconn);
200     ssize_t retval;
201
202     if (s->txbuf) {
203         return EAGAIN;
204     }
205
206     retval = stream_send(s->stream, buffer->data, buffer->size);
207     if (retval == buffer->size) {
208         ofpbuf_delete(buffer);
209         return 0;
210     } else if (retval >= 0 || retval == -EAGAIN) {
211         leak_checker_claim(buffer);
212         s->txbuf = buffer;
213         if (retval > 0) {
214             ofpbuf_pull(buffer, retval);
215         }
216         return 0;
217     } else {
218         return -retval;
219     }
220 }
221
222 static void
223 vconn_stream_run(struct vconn *vconn)
224 {
225     struct vconn_stream *s = vconn_stream_cast(vconn);
226     ssize_t retval;
227
228     if (!s->txbuf) {
229         return;
230     }
231
232     retval = stream_send(s->stream, s->txbuf->data, s->txbuf->size);
233     if (retval < 0) {
234         if (retval != -EAGAIN) {
235             VLOG_ERR_RL(&rl, "send: %s", strerror(-retval));
236             vconn_stream_clear_txbuf(s);
237             return;
238         }
239     } else if (retval > 0) {
240         ofpbuf_pull(s->txbuf, retval);
241         if (!s->txbuf->size) {
242             vconn_stream_clear_txbuf(s);
243             return;
244         }
245     }
246 }
247
248 static void
249 vconn_stream_run_wait(struct vconn *vconn)
250 {
251     struct vconn_stream *s = vconn_stream_cast(vconn);
252
253     if (s->txbuf) {
254         stream_send_wait(s->stream);
255     }
256 }
257
258 static void
259 vconn_stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
260 {
261     struct vconn_stream *s = vconn_stream_cast(vconn);
262     switch (wait) {
263     case WAIT_CONNECT:
264         stream_connect_wait(s->stream);
265         break;
266
267     case WAIT_SEND:
268         if (!s->txbuf) {
269             stream_send_wait(s->stream);
270         } else {
271             /* Nothing to do: need to drain txbuf first.
272              * vconn_stream_run_wait() will arrange to wake up when there room
273              * to send data, so there's no point in calling poll_fd_wait()
274              * redundantly here. */
275         }
276         break;
277
278     case WAIT_RECV:
279         stream_recv_wait(s->stream);
280         break;
281
282     default:
283         NOT_REACHED();
284     }
285 }
286 \f
287 /* Passive stream socket vconn. */
288
289 struct pvconn_pstream
290 {
291     struct pvconn pvconn;
292     struct pstream *pstream;
293 };
294
295 static struct pvconn_class pstream_pvconn_class;
296
297 static struct pvconn_pstream *
298 pvconn_pstream_cast(struct pvconn *pvconn)
299 {
300     return CONTAINER_OF(pvconn, struct pvconn_pstream, pvconn);
301 }
302
303 /* Creates a new pvconn named 'name' that will accept new connections using
304  * pstream_accept() and stores a pointer to the pvconn in '*pvconnp'.
305  *
306  * Returns 0 if successful, otherwise a positive errno value.  (The current
307  * implementation never fails.) */
308 static int
309 pvconn_pstream_listen(const char *name, char *suffix OVS_UNUSED,
310                       struct pvconn **pvconnp)
311 {
312     struct pvconn_pstream *ps;
313     struct pstream *pstream;
314     int error;
315
316     error = pstream_open_with_default_ports(name, OFP_TCP_PORT, OFP_SSL_PORT,
317                                             &pstream);
318     if (error) {
319         return error;
320     }
321
322     ps = xmalloc(sizeof *ps);
323     pvconn_init(&ps->pvconn, &pstream_pvconn_class, name);
324     ps->pstream = pstream;
325     *pvconnp = &ps->pvconn;
326     return 0;
327 }
328
329 static void
330 pvconn_pstream_close(struct pvconn *pvconn)
331 {
332     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
333     pstream_close(ps->pstream);
334     free(ps);
335 }
336
337 static int
338 pvconn_pstream_accept(struct pvconn *pvconn, struct vconn **new_vconnp)
339 {
340     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
341     struct stream *stream;
342     int error;
343
344     error = pstream_accept(ps->pstream, &stream);
345     if (error) {
346         if (error != EAGAIN) {
347             VLOG_DBG_RL(&rl, "%s: accept: %s",
348                         pstream_get_name(ps->pstream), strerror(error));
349         }
350         return error;
351     }
352
353     *new_vconnp = vconn_stream_new(stream, 0);
354     return 0;
355 }
356
357 static void
358 pvconn_pstream_wait(struct pvconn *pvconn)
359 {
360     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
361     pstream_wait(ps->pstream);
362 }
363 \f
364 /* Stream-based vconns and pvconns. */
365
366 #define DEFINE_VCONN_STREAM_CLASS(NAME)             \
367         struct vconn_class NAME##_vconn_class = {   \
368             #NAME,                                  \
369             vconn_stream_open,                      \
370             vconn_stream_close,                     \
371             vconn_stream_connect,                   \
372             vconn_stream_recv,                      \
373             vconn_stream_send,                      \
374             vconn_stream_run,                       \
375             vconn_stream_run_wait,                  \
376             vconn_stream_wait,                      \
377         };
378
379 #define DEFINE_PVCONN_STREAM_CLASS(NAME)            \
380         struct pvconn_class NAME##_pvconn_class = { \
381             #NAME,                                  \
382             pvconn_pstream_listen,                  \
383             pvconn_pstream_close,                   \
384             pvconn_pstream_accept,                  \
385             pvconn_pstream_wait                     \
386         };
387
388 static DEFINE_VCONN_STREAM_CLASS(stream);
389 static DEFINE_PVCONN_STREAM_CLASS(pstream);
390
391 DEFINE_VCONN_STREAM_CLASS(tcp);
392 DEFINE_PVCONN_STREAM_CLASS(ptcp);
393
394 DEFINE_VCONN_STREAM_CLASS(unix);
395 DEFINE_PVCONN_STREAM_CLASS(punix);
396
397 #ifdef HAVE_OPENSSL
398 DEFINE_VCONN_STREAM_CLASS(ssl);
399 DEFINE_PVCONN_STREAM_CLASS(pssl);
400 #endif