From 60cb3eb8b296e2aebbda6ccc161e99ad2bc7ca4a Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Wed, 6 Jan 2010 14:27:46 -0800 Subject: [PATCH] vconn: Convert vconn code to modern OVS structure. 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. --- lib/rconn.c | 21 +++++++++++++- lib/vconn-provider.h | 12 ++++++++ lib/vconn-ssl.c | 57 +++++++++++++++--------------------- lib/vconn-stream.c | 69 ++++++++++++++++++++++++++------------------ lib/vconn-tcp.c | 2 ++ lib/vconn-unix.c | 2 ++ lib/vconn.c | 31 ++++++++++++++++++-- lib/vconn.h | 3 ++ tests/test-vconn.c | 8 +++++ 9 files changed, 141 insertions(+), 64 deletions(-) diff --git a/lib/rconn.c b/lib/rconn.c index b6e958ee..f2d074aa 100644 --- a/lib/rconn.c +++ b/lib/rconn.c @@ -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()); diff --git a/lib/vconn-provider.h b/lib/vconn-provider.h index f245e4c1..1c8b86da 100644 --- a/lib/vconn-provider.h +++ b/lib/vconn-provider.h @@ -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); diff --git a/lib/vconn-ssl.c b/lib/vconn-ssl.c index 58c54f87..773090de 100644 --- a/lib/vconn-ssl.c +++ b/lib/vconn-ssl.c @@ -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 */ }; diff --git a/lib/vconn-stream.c b/lib/vconn-stream.c index 0551c9eb..f19f3ebf 100644 --- a/lib/vconn-stream.c +++ b/lib/vconn-stream.c @@ -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 */ }; diff --git a/lib/vconn-tcp.c b/lib/vconn-tcp.c index aac71662..e3a24f58 100644 --- a/lib/vconn-tcp.c +++ b/lib/vconn-tcp.c @@ -91,6 +91,8 @@ struct vconn_class tcp_vconn_class = { NULL, /* connect */ NULL, /* recv */ NULL, /* send */ + NULL, /* run */ + NULL, /* run_wait */ NULL, /* wait */ }; diff --git a/lib/vconn-unix.c b/lib/vconn-unix.c index f637ca0e..ff01022e 100644 --- a/lib/vconn-unix.c +++ b/lib/vconn-unix.c @@ -71,6 +71,8 @@ struct vconn_class unix_vconn_class = { NULL, /* connect */ NULL, /* recv */ NULL, /* send */ + NULL, /* run */ + NULL, /* run_wait */ NULL, /* wait */ }; diff --git a/lib/vconn.c b/lib/vconn.c index 7b9ae610..1445be48 100644 --- a/lib/vconn.c +++ b/lib/vconn.c @@ -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(); } diff --git a/lib/vconn.h b/lib/vconn.h index 0c13744c..9bd235ae 100644 --- a/lib/vconn.h +++ b/lib/vconn.h @@ -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 **); diff --git a/tests/test-vconn.c b/tests/test-vconn.c index 34c2930a..87e35c4a 100644 --- a/tests/test-vconn.c +++ b/tests/test-vconn.c @@ -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); } -- 2.30.2