mir/openvswitch
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-10-21 14:49:41 +00:00
Code Issues Releases Wiki Activity

vconn: Convert vconn code to modern OVS structure.

Browse Source 
The vconn code is a relative fossil as OVS code goes.  It was written
before we had really figured how code should fit together.  Part of that
history is that it used poll_fd_callback() to register callbacks without
the assistance of other code.  That isn't how the rest of OVS works now;
this code is the only remaining user of that function.

To make it more like the rest of the system, this code gets rid of the use
of poll_fd_callback().  It also adds vconn_run() and vconn_run_wait()
functions and calls to them from the places where they are now required.
This commit is contained in:
Ben Pfaff
2010-01-06 14:27:46 -08:00
parent 539e96f623
commit 60cb3eb8b2
9 changed files with 141 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
lib/rconn.c
View File

@@ -459,6 +459,15 @@ void
rconn_run(struct rconn *rc)
{
int old_state;
size_t i;
if (rc->vconn) {
vconn_run(rc->vconn);
}
for (i = 0; i < rc->n_monitors; i++) {
vconn_run(rc->monitors[i]);
}
do {
old_state = rc->state;
switch (rc->state) {
@@ -476,7 +485,17 @@ rconn_run(struct rconn *rc)
void
rconn_run_wait(struct rconn *rc)
{
unsigned int timeo = timeout(rc);
unsigned int timeo;
size_t i;
if (rc->vconn) {
vconn_run_wait(rc->vconn);
}
for (i = 0; i < rc->n_monitors; i++) {
vconn_run_wait(rc->monitors[i]);
}
timeo = timeout(rc);
if (timeo != UINT_MAX) {
unsigned int expires = sat_add(rc->state_entered, timeo);
unsigned int remaining = sat_sub(expires, time_now());

12
lib/vconn-provider.h
View File

@@ -108,6 +108,18 @@ struct vconn_class {
* accepted for transmission, it should return EAGAIN. */
int (*send)(struct vconn *vconn, struct ofpbuf *msg);
/* Allows 'vconn' to perform maintenance activities, such as flushing
* output buffers.
*
* May be null if 'vconn' doesn't have anything to do here. */
void (*run)(struct vconn *vconn);
/* Arranges for the poll loop to wake up when 'vconn' needs to perform
* maintenance activities.
*
* May be null if 'vconn' doesn't have anything to do here. */
void (*run_wait)(struct vconn *vconn);
/* Arranges for the poll loop to wake up when 'vconn' is ready to take an
* action of the given 'type'. */
void (*wait)(struct vconn *vconn, enum vconn_wait_type type);

57
lib/vconn-ssl.c
View File

@@ -67,7 +67,6 @@ struct ssl_vconn
SSL *ssl;
struct ofpbuf *rxbuf;
struct ofpbuf *txbuf;
struct poll_waiter *tx_waiter;
/* rx_want and tx_want record the result of the last call to SSL_read()
* and SSL_write(), respectively:
@@ -157,7 +156,6 @@ static void ssl_close(struct vconn *);
static void ssl_clear_txbuf(struct ssl_vconn *);
static int interpret_ssl_error(const char *function, int ret, int error,
int *want);
static void ssl_tx_poll_callback(int fd, short int revents, void *vconn_);
static DH *tmp_dh_callback(SSL *ssl, int is_export UNUSED, int keylength);
static void log_ca_cert(const char *file_name, X509 *cert);
@@ -257,7 +255,6 @@ new_ssl_vconn(const char *name, int fd, enum session_type type,
sslv->ssl = ssl;
sslv->rxbuf = NULL;
sslv->txbuf = NULL;
sslv->tx_waiter = NULL;
sslv->rx_want = sslv->tx_want = SSL_NOTHING;
*vconnp = &sslv->vconn;
return 0;
@@ -441,7 +438,6 @@ static void
ssl_close(struct vconn *vconn)
{
struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
poll_cancel(sslv->tx_waiter);
ssl_clear_txbuf(sslv);
ofpbuf_delete(sslv->rxbuf);
SSL_free(sslv->ssl);
@@ -565,10 +561,6 @@ again:
ret = SSL_read(sslv->ssl, ofpbuf_tail(rx), want_bytes);
if (old_state != SSL_get_state(sslv->ssl)) {
sslv->tx_want = SSL_NOTHING;
if (sslv->tx_waiter) {
poll_cancel(sslv->tx_waiter);
ssl_tx_poll_callback(sslv->fd, POLLIN, vconn);
}
}
sslv->rx_want = SSL_NOTHING;
@@ -605,16 +597,6 @@ ssl_clear_txbuf(struct ssl_vconn *sslv)
{
ofpbuf_delete(sslv->txbuf);
sslv->txbuf = NULL;
sslv->tx_waiter = NULL;
}
static void
ssl_register_tx_waiter(struct vconn *vconn)
{
struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
sslv->tx_waiter = poll_fd_callback(sslv->fd,
want_to_poll_events(sslv->tx_want),
ssl_tx_poll_callback, vconn);
}
static int
@@ -647,19 +629,6 @@ ssl_do_tx(struct vconn *vconn)
}
}
static void
ssl_tx_poll_callback(int fd UNUSED, short int revents UNUSED, void *vconn_)
{
struct vconn *vconn = vconn_;
struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
int error = ssl_do_tx(vconn);
if (error != EAGAIN) {
ssl_clear_txbuf(sslv);
} else {
ssl_register_tx_waiter(vconn);
}
}
static int
ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
{
@@ -678,7 +647,6 @@ ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
return 0;
case EAGAIN:
leak_checker_claim(buffer);
ssl_register_tx_waiter(vconn);
return 0;
default:
sslv->txbuf = NULL;
@@ -687,6 +655,26 @@ ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
}
}
static void
ssl_run(struct vconn *vconn)
{
struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
if (sslv->txbuf && ssl_do_tx(vconn) != EAGAIN) {
ssl_clear_txbuf(sslv);
}
}
static void
ssl_run_wait(struct vconn *vconn)
{
struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
if (sslv->tx_want != SSL_NOTHING) {
poll_fd_wait(sslv->fd, want_to_poll_events(sslv->tx_want));
}
}
static void
ssl_wait(struct vconn *vconn, enum vconn_wait_type wait)
{
@@ -728,7 +716,8 @@ ssl_wait(struct vconn *vconn, enum vconn_wait_type wait)
/* We have room in our tx queue. */
poll_immediate_wake();
} else {
/* The call to ssl_tx_poll_callback() will wake us up. */
/* vconn_run_wait() will do the right thing; don't bother with
* redundancy. */
}
break;
@@ -744,6 +733,8 @@ struct vconn_class ssl_vconn_class = {
ssl_connect, /* connect */
ssl_recv, /* recv */
ssl_send, /* send */
ssl_run, /* run */
ssl_run_wait, /* run_wait */
ssl_wait, /* wait */
};

69
lib/vconn-stream.c
View File

@@ -44,7 +44,6 @@ struct stream_vconn
int fd;
struct ofpbuf *rxbuf;
struct ofpbuf *txbuf;
struct poll_waiter *tx_waiter;
char *unlink_path;
};
@@ -74,7 +73,6 @@ new_stream_vconn(const char *name, int fd, int connect_status,
vconn_init(&s->vconn, &stream_vconn_class, connect_status, name);
s->fd = fd;
s->txbuf = NULL;
s->tx_waiter = NULL;
s->rxbuf = NULL;
s->unlink_path = unlink_path;
*vconnp = &s->vconn;
@@ -92,7 +90,6 @@ static void
stream_close(struct vconn *vconn)
{
struct stream_vconn *s = stream_vconn_cast(vconn);
poll_cancel(s->tx_waiter);
stream_clear_txbuf(s);
ofpbuf_delete(s->rxbuf);
close(s->fd);
@@ -170,29 +167,6 @@ stream_clear_txbuf(struct stream_vconn *s)
{
ofpbuf_delete(s->txbuf);
s->txbuf = NULL;
s->tx_waiter = NULL;
}
static void
stream_do_tx(int fd UNUSED, short int revents UNUSED, void *vconn_)
{
struct vconn *vconn = vconn_;
struct stream_vconn *s = stream_vconn_cast(vconn);
ssize_t n = write(s->fd, s->txbuf->data, s->txbuf->size);
if (n < 0) {
if (errno != EAGAIN) {
VLOG_ERR_RL(&rl, "send: %s", strerror(errno));
stream_clear_txbuf(s);
return;
}
} else if (n > 0) {
ofpbuf_pull(s->txbuf, n);
if (!s->txbuf->size) {
stream_clear_txbuf(s);
return;
}
}
s->tx_waiter = poll_fd_callback(s->fd, POLLOUT, stream_do_tx, vconn);
}
static int
@@ -215,13 +189,48 @@ stream_send(struct vconn *vconn, struct ofpbuf *buffer)
if (retval > 0) {
ofpbuf_pull(buffer, retval);
}
s->tx_waiter = poll_fd_callback(s->fd, POLLOUT, stream_do_tx, vconn);
return 0;
} else {
return errno;
}
}
static void
stream_run(struct vconn *vconn)
{
struct stream_vconn *s = stream_vconn_cast(vconn);
ssize_t n;
if (!s->txbuf) {
return;
}
n = write(s->fd, s->txbuf->data, s->txbuf->size);
if (n < 0) {
if (errno != EAGAIN) {
VLOG_ERR_RL(&rl, "send: %s", strerror(errno));
stream_clear_txbuf(s);
return;
}
} else if (n > 0) {
ofpbuf_pull(s->txbuf, n);
if (!s->txbuf->size) {
stream_clear_txbuf(s);
return;
}
}
}
static void
stream_run_wait(struct vconn *vconn)
{
struct stream_vconn *s = stream_vconn_cast(vconn);
if (s->txbuf) {
poll_fd_wait(s->fd, POLLOUT);
}
}
static void
stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
{
@@ -235,7 +244,9 @@ stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
if (!s->txbuf) {
poll_fd_wait(s->fd, POLLOUT);
} else {
/* Nothing to do: need to drain txbuf first. */
/* Nothing to do: need to drain txbuf first. stream_run_wait()
* will arrange to wake up when there room to send data, so there's
* no point in calling poll_fd_wait() redundantly here. */
}
break;
@@ -255,6 +266,8 @@ static struct vconn_class stream_vconn_class = {
stream_connect, /* connect */
stream_recv, /* recv */
stream_send, /* send */
stream_run, /* run */
stream_run_wait, /* run_wait */
stream_wait, /* wait */
};

2
lib/vconn-tcp.c
View File

@@ -91,6 +91,8 @@ struct vconn_class tcp_vconn_class = {
NULL, /* connect */
NULL, /* recv */
NULL, /* send */
NULL, /* run */
NULL, /* run_wait */
NULL, /* wait */
};

2
lib/vconn-unix.c
View File

@@ -71,6 +71,8 @@ struct vconn_class unix_vconn_class = {
NULL, /* connect */
NULL, /* recv */
NULL, /* send */
NULL, /* run */
NULL, /* run_wait */
NULL, /* wait */
};

31
lib/vconn.c
View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) 2008, 2009 Nicira Networks.
* Copyright (c) 2008, 2009, 2010 Nicira Networks.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -90,7 +90,8 @@ check_vconn_classes(void)
struct vconn_class *class = vconn_classes[i];
assert(class->name != NULL);
assert(class->open != NULL);
if (class->close || class->recv || class->send || class->wait) {
if (class->close || class->recv || class->send
|| class->run || class->run_wait || class->wait) {
assert(class->close != NULL);
assert(class->recv != NULL);
assert(class->send != NULL);
@@ -208,6 +209,26 @@ vconn_open(const char *name, int min_version, struct vconn **vconnp)
return EAFNOSUPPORT;
}
/* Allows 'vconn' to perform maintenance activities, such as flushing output
* buffers. */
void
vconn_run(struct vconn *vconn)
{
if (vconn->class->run) {
(vconn->class->run)(vconn);
}
}
/* Arranges for the poll loop to wake up when 'vconn' needs to perform
* maintenance activities. */
void
vconn_run_wait(struct vconn *vconn)
{
if (vconn->class->run_wait) {
(vconn->class->run_wait)(vconn);
}
}
int
vconn_open_block(const char *name, int min_version, struct vconn **vconnp)
{
@@ -216,6 +237,8 @@ vconn_open_block(const char *name, int min_version, struct vconn **vconnp)
error = vconn_open(name, min_version, &vconn);
while (error == EAGAIN) {
vconn_run(vconn);
vconn_run_wait(vconn);
vconn_connect_wait(vconn);
poll_block();
error = vconn_connect(vconn);
@@ -547,6 +570,8 @@ vconn_send_block(struct vconn *vconn, struct ofpbuf *msg)
{
int retval;
while ((retval = vconn_send(vconn, msg)) == EAGAIN) {
vconn_run(vconn);
vconn_run_wait(vconn);
vconn_send_wait(vconn);
poll_block();
}
@@ -559,6 +584,8 @@ vconn_recv_block(struct vconn *vconn, struct ofpbuf **msgp)
{
int retval;
while ((retval = vconn_recv(vconn, msgp)) == EAGAIN) {
vconn_run(vconn);
vconn_run_wait(vconn);
vconn_recv_wait(vconn);
poll_block();
}

3
lib/vconn.h
View File

@@ -48,6 +48,9 @@ int vconn_send(struct vconn *, struct ofpbuf *);
int vconn_recv_xid(struct vconn *, uint32_t xid, struct ofpbuf **);
int vconn_transact(struct vconn *, struct ofpbuf *, struct ofpbuf **);
void vconn_run(struct vconn *);
void vconn_run_wait(struct vconn *);
int vconn_open_block(const char *name, int min_version, struct vconn **);
int vconn_send_block(struct vconn *, struct ofpbuf *);
int vconn_recv_block(struct vconn *, struct ofpbuf **);

8
tests/test-vconn.c
View File

@@ -143,6 +143,7 @@ test_refuse_connection(const char *type, int expected_error)
fpv_create(type, &fpv);
assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
fpv_close(&fpv);
vconn_run(vconn);
assert(vconn_connect(vconn) == expected_error);
vconn_close(vconn);
fpv_destroy(&fpv);
@@ -159,6 +160,7 @@ test_accept_then_close(const char *type, int expected_error)
fpv_create(type, &fpv);
assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
vconn_run(vconn);
close(fpv_accept(&fpv));
fpv_close(&fpv);
assert(vconn_connect(vconn) == expected_error);
@@ -178,6 +180,7 @@ test_read_hello(const char *type, int expected_error)
fpv_create(type, &fpv);
assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
vconn_run(vconn);
fd = fpv_accept(&fpv);
fpv_destroy(&fpv);
assert(!set_nonblocking(fd));
@@ -195,7 +198,9 @@ test_read_hello(const char *type, int expected_error)
assert(errno == EAGAIN);
}
vconn_run(vconn);
assert(vconn_connect(vconn) == EAGAIN);
vconn_run_wait(vconn);
vconn_connect_wait(vconn);
poll_fd_wait(fd, POLLIN);
poll_block();
@@ -221,6 +226,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
fpv_create(type, &fpv);
assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
vconn_run(vconn);
fd = fpv_accept(&fpv);
fpv_destroy(&fpv);
@@ -243,6 +249,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
}
}
vconn_run(vconn);
if (!connected) {
int error = vconn_connect(vconn);
if (error == expect_connect_error) {
@@ -262,6 +269,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
break;
}
vconn_run_wait(vconn);
if (!connected) {
vconn_connect_wait(vconn);
}