[ovs-dev] [PATCH 3/9] stream: Add stream_run(), stream_run_wait() functions.
Ben Pfaff
blp at nicira.com
Mon Dec 21 21:15:43 UTC 2009
SSL, which will be added in an upcoming commit, requires some background
processing, which is best done in a "run" function in our architecture.
This commit adds stream_run() and stream_run_wait() and calls to them from
the places where they will be required.
---
lib/jsonrpc.c | 8 +++++++-
lib/stream-fd.c | 2 ++
lib/stream-provider.h | 12 ++++++++++++
lib/stream-tcp.c | 2 ++
lib/stream-unix.c | 2 ++
lib/stream.c | 29 ++++++++++++++++++++++++++++-
lib/stream.h | 3 +++
ovsdb/jsonrpc-server.c | 6 +++++-
8 files changed, 61 insertions(+), 3 deletions(-)
diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c
index bd019f7..81a2ef6 100644
--- a/lib/jsonrpc.c
+++ b/lib/jsonrpc.c
@@ -88,6 +88,7 @@ jsonrpc_run(struct jsonrpc *rpc)
return;
}
+ stream_run(rpc->stream);
while (!queue_is_empty(&rpc->output)) {
struct ofpbuf *buf = rpc->output.head;
int retval;
@@ -113,6 +114,7 @@ jsonrpc_run(struct jsonrpc *rpc)
void
jsonrpc_wait(struct jsonrpc *rpc)
{
+ stream_run_wait(rpc->stream);
if (!rpc->status && !queue_is_empty(&rpc->output)) {
stream_send_wait(rpc->stream);
}
@@ -699,7 +701,10 @@ jsonrpc_session_run(struct jsonrpc_session *s)
jsonrpc_session_disconnect(s);
}
} else if (s->stream) {
- int error = stream_connect(s->stream);
+ int error;
+
+ stream_run(s->stream);
+ error = stream_connect(s->stream);
if (!error) {
reconnect_connected(s->reconnect, time_msec());
s->rpc = jsonrpc_open(s->stream);
@@ -741,6 +746,7 @@ jsonrpc_session_wait(struct jsonrpc_session *s)
if (s->rpc) {
jsonrpc_wait(s->rpc);
} else if (s->stream) {
+ stream_run_wait(s->stream);
stream_connect_wait(s->stream);
}
reconnect_wait(s->reconnect, time_msec());
diff --git a/lib/stream-fd.c b/lib/stream-fd.c
index 46aa8e7..94c8434 100644
--- a/lib/stream-fd.c
+++ b/lib/stream-fd.c
@@ -139,6 +139,8 @@ static struct stream_class stream_fd_class = {
fd_connect, /* connect */
fd_recv, /* recv */
fd_send, /* send */
+ NULL, /* run */
+ NULL, /* run_wait */
fd_wait, /* wait */
};
diff --git a/lib/stream-provider.h b/lib/stream-provider.h
index 6beaab7..872da3c 100644
--- a/lib/stream-provider.h
+++ b/lib/stream-provider.h
@@ -109,6 +109,18 @@ struct stream_class {
* accepted for transmission, it should return -EAGAIN immediately. */
ssize_t (*send)(struct stream *stream, const void *buffer, size_t n);
+ /* Allows 'stream' to perform maintenance activities, such as flushing
+ * output buffers.
+ *
+ * May be null if 'stream' doesn't have anything to do here. */
+ void (*run)(struct stream *stream);
+
+ /* Arranges for the poll loop to wake up when 'stream' needs to perform
+ * maintenance activities.
+ *
+ * May be null if 'stream' doesn't have anything to do here. */
+ void (*run_wait)(struct stream *stream);
+
/* Arranges for the poll loop to wake up when 'stream' is ready to take an
* action of the given 'type'. */
void (*wait)(struct stream *stream, enum stream_wait_type type);
diff --git a/lib/stream-tcp.c b/lib/stream-tcp.c
index 947be9f..bfcf35c 100644
--- a/lib/stream-tcp.c
+++ b/lib/stream-tcp.c
@@ -90,6 +90,8 @@ struct stream_class tcp_stream_class = {
NULL, /* connect */
NULL, /* recv */
NULL, /* send */
+ NULL, /* run */
+ NULL, /* run_wait */
NULL, /* wait */
};
diff --git a/lib/stream-unix.c b/lib/stream-unix.c
index 9046da1..6ce7790 100644
--- a/lib/stream-unix.c
+++ b/lib/stream-unix.c
@@ -69,6 +69,8 @@ struct stream_class unix_stream_class = {
NULL, /* connect */
NULL, /* recv */
NULL, /* send */
+ NULL, /* run */
+ NULL, /* run_wait */
NULL, /* wait */
};
diff --git a/lib/stream.c b/lib/stream.c
index 337fb5c..7511981 100644
--- a/lib/stream.c
+++ b/lib/stream.c
@@ -66,10 +66,13 @@ check_stream_classes(void)
struct stream_class *class = stream_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);
+ assert(class->run != NULL);
+ assert(class->run_wait != NULL);
assert(class->wait != NULL);
} else {
/* This class delegates to another one. */
@@ -166,6 +169,8 @@ stream_open_block(const char *name, struct stream **streamp)
error = stream_open(name, &stream);
while (error == EAGAIN) {
+ stream_run(stream);
+ stream_run_wait(stream);
stream_connect_wait(stream);
poll_block();
error = stream_connect(stream);
@@ -312,6 +317,28 @@ stream_send(struct stream *stream, const void *buffer, size_t n)
: (stream->class->send)(stream, buffer, n));
}
+/* Allows 'stream' to perform maintenance activities, such as flushing
+ * output buffers. */
+void
+stream_run(struct stream *stream)
+{
+ if (stream->class->run) {
+ (stream->class->run)(stream);
+ }
+}
+
+/* Arranges for the poll loop to wake up when 'stream' needs to perform
+ * maintenance activities. */
+void
+stream_run_wait(struct stream *stream)
+{
+ if (stream->class->run_wait) {
+ (stream->class->run_wait)(stream);
+ }
+}
+
+/* Arranges for the poll loop to wake up when 'stream' is ready to take an
+ * action of the given 'type'. */
void
stream_wait(struct stream *stream, enum stream_wait_type wait)
{
diff --git a/lib/stream.h b/lib/stream.h
index 7a62a5a..ae30b10 100644
--- a/lib/stream.h
+++ b/lib/stream.h
@@ -42,6 +42,9 @@ int stream_connect(struct stream *);
int stream_recv(struct stream *, void *buffer, size_t n);
int stream_send(struct stream *, const void *buffer, size_t n);
+void stream_run(struct stream *);
+void stream_run_wait(struct stream *);
+
enum stream_wait_type {
STREAM_CONNECT,
STREAM_RECV,
diff --git a/ovsdb/jsonrpc-server.c b/ovsdb/jsonrpc-server.c
index a8e724a..d7d9b23 100644
--- a/ovsdb/jsonrpc-server.c
+++ b/ovsdb/jsonrpc-server.c
@@ -293,7 +293,10 @@ ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *s)
}
}
} else if (s->stream) {
- int error = stream_connect(s->stream);
+ int error;
+
+ stream_run(s->stream);
+ error = stream_connect(s->stream);
if (!error) {
reconnect_connected(s->reconnect, time_msec());
s->rpc = jsonrpc_open(s->stream);
@@ -353,6 +356,7 @@ ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *s)
jsonrpc_recv_wait(s->rpc);
}
} else if (s->stream) {
+ stream_run_wait(s->stream);
stream_connect_wait(s->stream);
}
reconnect_wait(s->reconnect, time_msec());
--
1.6.3.3
More information about the dev
mailing list