diff mbox series

[ovs-dev,v2,2/9] stream: Add record/replay functionality.

Message ID 20210412220020.2286954-3-i.maximets@ovn.org
State Changes Requested
Headers show
Series Stream Record/Replay. | expand

Commit Message

Ilya Maximets April 12, 2021, 10 p.m. UTC 
For debugging purposes it is useful to be able to record all the
incoming transactions and commands and replay them locally under
debugger or with additional logging enabled.  This patch introduces
ability to record all the incoming stream data and replay it via new
stream provider named 'stream-replay'.  During the record phase all
the incoming stream data written to special replay_* files in the
application rundir.  On replay phase instead of opening real streams
application will open replay_* files and read all the incoming data
directly from them.

If enabled for ovsdb-server, for example, this allows to record all
the connections and transactions from the big setup and replay them
locally afterwards to debug the behaviour or test performance.

To start application in recording mode there is a --replay-record
cmdline option. --replay is to replay previously recorded streams.

Current version doesn't work well with time-based stream events like
inactivity probes or any other events generated internally.  This is
a point for further improvement.

Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
---
 lib/automake.mk       |   1 +
 lib/stream-provider.h |   5 +
 lib/stream-replay.c   | 459 ++++++++++++++++++++++++++++++++++++++++++
 lib/stream.c          |  35 +++-
 lib/stream.h          |  12 ++
 5 files changed, 506 insertions(+), 6 deletions(-)
 create mode 100644 lib/stream-replay.c

Comments

Dumitru Ceara May 10, 2021, 10:13 a.m. UTC | #1 
On 4/13/21 12:00 AM, Ilya Maximets wrote:
> For debugging purposes it is useful to be able to record all the
> incoming transactions and commands and replay them locally under
> debugger or with additional logging enabled.  This patch introduces
> ability to record all the incoming stream data and replay it via new
> stream provider named 'stream-replay'.  During the record phase all
> the incoming stream data written to special replay_* files in the
> application rundir.  On replay phase instead of opening real streams
> application will open replay_* files and read all the incoming data
> directly from them.
> 
> If enabled for ovsdb-server, for example, this allows to record all
> the connections and transactions from the big setup and replay them
> locally afterwards to debug the behaviour or test performance.
> 
> To start application in recording mode there is a --replay-record
> cmdline option. --replay is to replay previously recorded streams.
> 
> Current version doesn't work well with time-based stream events like
> inactivity probes or any other events generated internally.  This is
> a point for further improvement.
> 
> Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
> ---
>  lib/automake.mk       |   1 +
>  lib/stream-provider.h |   5 +
>  lib/stream-replay.c   | 459 ++++++++++++++++++++++++++++++++++++++++++
>  lib/stream.c          |  35 +++-
>  lib/stream.h          |  12 ++
>  5 files changed, 506 insertions(+), 6 deletions(-)
>  create mode 100644 lib/stream-replay.c
> 
> diff --git a/lib/automake.mk b/lib/automake.mk
> index b558692c6..db9017591 100644
> --- a/lib/automake.mk
> +++ b/lib/automake.mk
> @@ -312,6 +312,7 @@ lib_libopenvswitch_la_SOURCES = \
>  	lib/stream-fd.c \
>  	lib/stream-fd.h \
>  	lib/stream-provider.h \
> +	lib/stream-replay.c \
>  	lib/stream-ssl.h \
>  	lib/stream-tcp.c \
>  	lib/stream.c \
> diff --git a/lib/stream-provider.h b/lib/stream-provider.h
> index 75f4f059b..44e3c6431 100644
> --- a/lib/stream-provider.h
> +++ b/lib/stream-provider.h
> @@ -18,6 +18,7 @@
>  #define STREAM_PROVIDER_H 1
>  
>  #include <sys/types.h>
> +#include "ovs-replay.h"
>  #include "stream.h"
>  
>  /* Active stream connection. */
> @@ -29,6 +30,7 @@ struct stream {
>      const struct stream_class *class;
>      int state;
>      int error;
> +    replay_file_t replay_wfd;
>      char *name;
>      char *peer_id;
>  };
> @@ -133,6 +135,7 @@ struct pstream {
>      const struct pstream_class *class;
>      char *name;
>      ovs_be16 bound_port;
> +    replay_file_t replay_wfd;
>  };
>  
>  void pstream_init(struct pstream *, const struct pstream_class *, char *name);
> @@ -200,5 +203,7 @@ extern const struct pstream_class pwindows_pstream_class;
>  extern const struct stream_class ssl_stream_class;
>  extern const struct pstream_class pssl_pstream_class;
>  #endif
> +extern const struct stream_class replay_stream_class;
> +extern const struct pstream_class preplay_pstream_class;
>  
>  #endif /* stream-provider.h */
> diff --git a/lib/stream-replay.c b/lib/stream-replay.c
> new file mode 100644
> index 000000000..ef591b920
> --- /dev/null
> +++ b/lib/stream-replay.c
> @@ -0,0 +1,459 @@
> +/*
> + * Copyright (c) 2021, Red Hat, Inc.
> + *
> + * Licensed under the Apache License, Version 2.0 (the "License");
> + * you may not use this file except in compliance with the License.
> + * You may obtain a copy of the License at:
> + *
> + *     http://www.apache.org/licenses/LICENSE-2.0
> + *
> + * Unless required by applicable law or agreed to in writing, software
> + * distributed under the License is distributed on an "AS IS" BASIS,
> + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> + * See the License for the specific language governing permissions and
> + * limitations under the License.
> + */
> +
> +#include <config.h>
> +#include <ctype.h>
> +#include <errno.h>
> +#include <poll.h>
> +#include <stdlib.h>
> +#include <string.h>
> +#include <sys/socket.h>
> +#include <sys/types.h>
> +#include <unistd.h>
> +#include "ovs-atomic.h"
> +#include "ovs-replay.h"
> +#include "util.h"
> +#include "stream-provider.h"
> +#include "stream.h"
> +#include "openvswitch/poll-loop.h"
> +#include "openvswitch/vlog.h"
> +
> +VLOG_DEFINE_THIS_MODULE(stream_replay);
> +
> +/* Active replay stream. */
> +
> +struct stream_replay
> +{

Nit: Curly brace on the line above, please.

> +    struct stream stream;
> +    replay_file_t f;
> +    int seqno;
> +};
> +
> +const struct stream_class replay_stream_class;
> +
> +/* Creates a new stream named 'name' that will emulate sending and receiving
> + * data using replay file and stores a pointer to the stream in '*streamp'.
> + *
> + * Takes ownership of 'name'.
> + *
> + * Returns 0 if successful, otherwise a positive errno value. */
> +static int
> +new_replay_stream(char *name, struct stream **streamp)
> +{
> +    struct stream_replay *s;
> +    int seqno = 0, error = 0, open_result;
> +    replay_file_t f;
> +
> +    ovs_replay_lock();
> +    error = ovs_replay_file_open(name, &f, &seqno);
> +    if (error) {
> +        VLOG_ERR("%s: failed to open stream.", name);
> +        goto unlock;

I didn't try this out but don't we leak 'name' here?

> +    }
> +
> +    error = ovs_replay_read(f, NULL, 0, &open_result, &seqno, true);
> +    if (error) {
> +        VLOG_ERR("%s: failed to read 'open' record.", name);
> +        ovs_replay_file_close(f);
> +        goto unlock;

Same here.

> +    }
> +
> +    if (open_result) {
> +        error = -open_result;
> +        ovs_replay_file_close(f);
> +        goto unlock;

Here too.

> +    }
> +
> +    s = xmalloc(sizeof *s);
> +    stream_init(&s->stream, &replay_stream_class, 0, name);
> +    s->f = f;
> +    s->seqno = seqno;
> +    *streamp = &s->stream;
> +unlock:
> +    ovs_replay_unlock();
> +    return error;
> +}
> +
> +static struct stream_replay *
> +stream_replay_cast(struct stream *stream)
> +{
> +    stream_assert_class(stream, &replay_stream_class);
> +    return CONTAINER_OF(stream, struct stream_replay, stream);
> +}
> +
> +void
> +stream_replay_open_wfd(struct stream *s, int open_result, const char *name)
> +{
> +    int state = ovs_replay_get_state();
> +    int error = 0;
> +    replay_file_t f;
> +
> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
> +        return;
> +    }
> +
> +    ovs_replay_lock();
> +    error = ovs_replay_file_open(name, &f, NULL);
> +    if (error) {
> +        VLOG_ERR("%s: failed to open replay file for stream.", name);
> +        ovs_replay_unlock();
> +        return;
> +    }
> +    ovs_replay_unlock();
> +
> +    if (ovs_replay_write(f, NULL, -open_result, true)) {
> +        VLOG_ERR("%s: failed to write 'open' failure: %d",
> +                 s->name, open_result);
> +    }
> +    if (open_result) {
> +        /* We recorded failure to open the stream. */
> +        ovs_replay_file_close(f);
> +    } else {
> +        s->replay_wfd = f;
> +    }
> +}
> +
> +void
> +stream_replay_write(struct stream *s, const void *buffer, int n, bool is_read)
> +{
> +    int state = ovs_replay_get_state();
> +
> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
> +        return;
> +    }
> +
> +    if (ovs_replay_write(s->replay_wfd, buffer, n, is_read)) {
> +        VLOG_ERR("%s: failed to write buffer.", s->name);

ovs_replay_write() rate limits error logs, should we do that here too?
AFAICT this applies to all VLOG_ERR() calls in this file.

> +    }
> +}
> +
> +void
> +stream_replay_close_wfd(struct stream *s)
> +{
> +    if (s->replay_wfd) {
> +        ovs_replay_file_close(s->replay_wfd);
> +    }
> +}
> +
> +static int
> +stream_replay_open(const char *name, char *suffix OVS_UNUSED,
> +                   struct stream **streamp, uint8_t dscp OVS_UNUSED)
> +{
> +    return new_replay_stream(xstrdup(name), streamp);
> +}
> +
> +static void
> +stream_replay_close(struct stream *stream)
> +{
> +    struct stream_replay *s = stream_replay_cast(stream);
> +    ovs_replay_file_close(s->f);
> +    free(s);
> +}
> +
> +static ssize_t
> +stream_replay_recv(struct stream *stream, void *buffer, size_t n)
> +{
> +    struct stream_replay *s = stream_replay_cast(stream);
> +    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
> +    int error, len;
> +
> +    ovs_replay_lock();
> +    ovs_assert(norm_seqno >= ovs_replay_seqno());
> +
> +    if (norm_seqno != ovs_replay_seqno()
> +        || !ovs_replay_seqno_is_read(s->seqno)) {
> +        error = EAGAIN;
> +        goto unlock;
> +    }
> +
> +    error = ovs_replay_read(s->f, buffer, n, &len, &s->seqno, true);
> +    if (error) {
> +        VLOG_ERR("%s: failed to read from replay file.", stream->name);
> +        goto unlock;
> +    }
> +
> +unlock:
> +    ovs_replay_unlock();
> +    return error ? -error : len;
> +}
> +
> +static ssize_t
> +stream_replay_send(struct stream *stream OVS_UNUSED,
> +                   const void *buffer OVS_UNUSED, size_t n)
> +{
> +    struct stream_replay *s = stream_replay_cast(stream);
> +    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
> +    int error, len;
> +
> +    ovs_replay_lock();
> +    ovs_assert(norm_seqno >= ovs_replay_seqno());
> +
> +    if (norm_seqno != ovs_replay_seqno()
> +        || ovs_replay_seqno_is_read(s->seqno)) {
> +        error = EAGAIN;
> +        goto unlock;
> +    }
> +
> +    error = ovs_replay_read(s->f, NULL, 0, &len, &s->seqno, false);
> +    if (error) {
> +        VLOG_ERR("%s: failed to read from replay file.", stream->name);
> +        goto unlock;
> +    }
> +    ovs_assert(len < 0 || len <= n);
> +
> +unlock:
> +    ovs_replay_unlock();
> +    return error ? -error : len;
> +}
> +
> +static void
> +stream_replay_wait(struct stream *stream, enum stream_wait_type wait)
> +{
> +    struct stream_replay *s = stream_replay_cast(stream);
> +    switch (wait) {
> +    case STREAM_CONNECT:
> +        /* Connect does nothing and always available. */
> +        poll_immediate_wake();
> +        break;
> +
> +    case STREAM_SEND:
> +        if (s->seqno != INT_MAX && !ovs_replay_seqno_is_read(s->seqno)) {
> +            /* Stream waits for write. */
> +            poll_immediate_wake();
> +        }
> +        break;
> +
> +    case STREAM_RECV:
> +        if (s->seqno != INT_MAX && ovs_replay_seqno_is_read(s->seqno)) {
> +            /* We still have something to read. */
> +            poll_immediate_wake();
> +        }
> +        break;
> +
> +    default:
> +        OVS_NOT_REACHED();
> +    }
> +}
> +
> +const struct stream_class replay_stream_class = {
> +    "replay",                   /* name */
> +    false,                      /* needs_probes */
> +    stream_replay_open,         /* open */
> +    stream_replay_close,        /* close */
> +    NULL,                       /* connect */
> +    stream_replay_recv,         /* recv */
> +    stream_replay_send,         /* send */
> +    NULL,                       /* run */
> +    NULL,                       /* run_wait */
> +    stream_replay_wait,         /* wait */
> +};
> +
> +/* Passive replay stream. */
> +
> +struct replay_pstream
> +{
> +    struct pstream pstream;
> +    replay_file_t f;
> +    int seqno;
> +};
> +
> +const struct pstream_class preplay_pstream_class;
> +
> +static struct replay_pstream *
> +replay_pstream_cast(struct pstream *pstream)
> +{
> +    pstream_assert_class(pstream, &preplay_pstream_class);
> +    return CONTAINER_OF(pstream, struct replay_pstream, pstream);
> +}
> +
> +/* Creates a new pstream named 'name' that will accept new replay connections
> + * reading them from the replay file and stores a pointer to the stream in
> + * '*pstreamp'.
> + *
> + * Takes ownership of 'name'.
> + *
> + * Returns 0 if successful, otherwise a positive errno value. */
> +static int
> +pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
> +                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
> +{
> +    int seqno = 0, error = 0, listen_result;
> +    replay_file_t f;
> +
> +    ovs_replay_lock();
> +    error = ovs_replay_file_open(name, &f, &seqno);
> +    if (error) {
> +        VLOG_ERR("%s: failed to open pstream.", name);> +        goto unlock;

'name' is leaked here.

> +    }
> +
> +    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
> +    if (error) {
> +        VLOG_ERR("%s: failed to read 'listen' record.", name);
> +        ovs_replay_file_close(f);
> +        goto unlock;

Same here.

> +    }
> +
> +    if (listen_result) {
> +        error = -listen_result;
> +        ovs_replay_file_close(f);
> +        goto unlock;

Here too.

> +    }
> +
> +    struct replay_pstream *ps = xmalloc(sizeof *ps);
> +    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));

I guess the xstrdup(name) is not needed, otherwise we don't really take
ownership of 'name' and we actually leak it.

> +    ps->f = f;
> +    ps->seqno = seqno;
> +    *pstreamp = &ps->pstream;
> +unlock:
> +    ovs_replay_unlock();
> +    return error;
> +}
> +
> +void
> +pstream_replay_open_wfd(struct pstream *ps, int listen_result,
> +                        const char *name)
> +{
> +    int state = ovs_replay_get_state();
> +    int error = 0;
> +    replay_file_t f;
> +
> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
> +        return;
> +    }
> +
> +    ovs_replay_lock();
> +    error = ovs_replay_file_open(name, &f, NULL);
> +    if (error) {
> +        VLOG_ERR("%s: failed to open replay file for pstream.", name);
> +        ovs_replay_unlock();
> +        return;
> +    }
> +    ovs_replay_unlock();
> +
> +    if (ovs_replay_write(f, NULL, -listen_result, true)) {
> +        VLOG_ERR("%s: failed to write 'listen' result: %d",
> +                 ps->name, listen_result);
> +    }
> +
> +    if (listen_result) {
> +        /* We recorded failure to open the stream. */
> +        ovs_replay_file_close(f);
> +    } else {
> +        ps->replay_wfd = f;
> +    }
> +}
> +
> +
> +void
> +pstream_replay_write_accept(struct pstream *ps, const struct stream *s,
> +                            int accept_result)
> +{
> +    int state = ovs_replay_get_state();
> +    int len;
> +
> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
> +        return;
> +    }
> +
> +    if (!accept_result) {
> +        len = strlen(s->name);
> +        if (ovs_replay_write(ps->replay_wfd, s->name, len, true)) {
> +            VLOG_ERR("%s: failed to write accept name: %s", ps->name, s->name);
> +        }
> +    } else if (ovs_replay_write(ps->replay_wfd, NULL, -accept_result, true)) {
> +        VLOG_ERR("%s: failed to write 'accept' failure: %d",
> +                 ps->name, accept_result);
> +    }
> +}
> +
> +void
> +pstream_replay_close_wfd(struct pstream *ps)
> +{
> +    if (ps->replay_wfd) {
> +        ovs_replay_file_close(ps->replay_wfd);
> +    }
> +}
> +
> +

Nit: I'd remove the second newline.

Thanks,
Dumitru
Dumitru Ceara May 10, 2021, 7:04 p.m. UTC | #2 
On 5/10/21 12:13 PM, Dumitru Ceara wrote:
>> +/* Creates a new pstream named 'name' that will accept new replay connections
>> + * reading them from the replay file and stores a pointer to the stream in
>> + * '*pstreamp'.
>> + *
>> + * Takes ownership of 'name'.
>> + *
>> + * Returns 0 if successful, otherwise a positive errno value. */
>> +static int
>> +pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
>> +                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
>> +{
>> +    int seqno = 0, error = 0, listen_result;
>> +    replay_file_t f;
>> +
>> +    ovs_replay_lock();
>> +    error = ovs_replay_file_open(name, &f, &seqno);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to open pstream.", name);> +        goto unlock;
> 'name' is leaked here.
> 
>> +    }
>> +
>> +    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to read 'listen' record.", name);
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> Same here.
> 
>> +    }
>> +
>> +    if (listen_result) {
>> +        error = -listen_result;
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> Here too.
> 
>> +    }
>> +
>> +    struct replay_pstream *ps = xmalloc(sizeof *ps);
>> +    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));
> I guess the xstrdup(name) is not needed, otherwise we don't really take
> ownership of 'name' and we actually leak it.
> 

Actually, this is wrong, sorry for the noise.  'name' is not leaked, but
the function comment is misleading.
Ilya Maximets May 27, 2021, 12:43 p.m. UTC | #3 
On 5/10/21 9:04 PM, Dumitru Ceara wrote:
> On 5/10/21 12:13 PM, Dumitru Ceara wrote:
>>> +/* Creates a new pstream named 'name' that will accept new replay connections
>>> + * reading them from the replay file and stores a pointer to the stream in
>>> + * '*pstreamp'.
>>> + *
>>> + * Takes ownership of 'name'.
>>> + *
>>> + * Returns 0 if successful, otherwise a positive errno value. */
>>> +static int
>>> +pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
>>> +                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
>>> +{
>>> +    int seqno = 0, error = 0, listen_result;
>>> +    replay_file_t f;
>>> +
>>> +    ovs_replay_lock();
>>> +    error = ovs_replay_file_open(name, &f, &seqno);
>>> +    if (error) {
>>> +        VLOG_ERR("%s: failed to open pstream.", name);> +        goto unlock;
>> 'name' is leaked here.
>>
>>> +    }
>>> +
>>> +    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
>>> +    if (error) {
>>> +        VLOG_ERR("%s: failed to read 'listen' record.", name);
>>> +        ovs_replay_file_close(f);
>>> +        goto unlock;
>> Same here.
>>
>>> +    }
>>> +
>>> +    if (listen_result) {
>>> +        error = -listen_result;
>>> +        ovs_replay_file_close(f);
>>> +        goto unlock;
>> Here too.
>>
>>> +    }
>>> +
>>> +    struct replay_pstream *ps = xmalloc(sizeof *ps);
>>> +    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));
>> I guess the xstrdup(name) is not needed, otherwise we don't really take
>> ownership of 'name' and we actually leak it.
>>
> 
> Actually, this is wrong, sorry for the noise.  'name' is not leaked, but
> the function comment is misleading.
> 

Good catch!  I removed the incorrect comment.
Ilya Maximets May 27, 2021, 1:01 p.m. UTC | #4 
On 5/10/21 12:13 PM, Dumitru Ceara wrote:
> On 4/13/21 12:00 AM, Ilya Maximets wrote:
>> For debugging purposes it is useful to be able to record all the
>> incoming transactions and commands and replay them locally under
>> debugger or with additional logging enabled.  This patch introduces
>> ability to record all the incoming stream data and replay it via new
>> stream provider named 'stream-replay'.  During the record phase all
>> the incoming stream data written to special replay_* files in the
>> application rundir.  On replay phase instead of opening real streams
>> application will open replay_* files and read all the incoming data
>> directly from them.
>>
>> If enabled for ovsdb-server, for example, this allows to record all
>> the connections and transactions from the big setup and replay them
>> locally afterwards to debug the behaviour or test performance.
>>
>> To start application in recording mode there is a --replay-record
>> cmdline option. --replay is to replay previously recorded streams.
>>
>> Current version doesn't work well with time-based stream events like
>> inactivity probes or any other events generated internally.  This is
>> a point for further improvement.
>>
>> Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
>> ---
>>  lib/automake.mk       |   1 +
>>  lib/stream-provider.h |   5 +
>>  lib/stream-replay.c   | 459 ++++++++++++++++++++++++++++++++++++++++++
>>  lib/stream.c          |  35 +++-
>>  lib/stream.h          |  12 ++
>>  5 files changed, 506 insertions(+), 6 deletions(-)
>>  create mode 100644 lib/stream-replay.c
>>
>> diff --git a/lib/automake.mk b/lib/automake.mk
>> index b558692c6..db9017591 100644
>> --- a/lib/automake.mk
>> +++ b/lib/automake.mk
>> @@ -312,6 +312,7 @@ lib_libopenvswitch_la_SOURCES = \
>>  	lib/stream-fd.c \
>>  	lib/stream-fd.h \
>>  	lib/stream-provider.h \
>> +	lib/stream-replay.c \
>>  	lib/stream-ssl.h \
>>  	lib/stream-tcp.c \
>>  	lib/stream.c \
>> diff --git a/lib/stream-provider.h b/lib/stream-provider.h
>> index 75f4f059b..44e3c6431 100644
>> --- a/lib/stream-provider.h
>> +++ b/lib/stream-provider.h
>> @@ -18,6 +18,7 @@
>>  #define STREAM_PROVIDER_H 1
>>  
>>  #include <sys/types.h>
>> +#include "ovs-replay.h"
>>  #include "stream.h"
>>  
>>  /* Active stream connection. */
>> @@ -29,6 +30,7 @@ struct stream {
>>      const struct stream_class *class;
>>      int state;
>>      int error;
>> +    replay_file_t replay_wfd;
>>      char *name;
>>      char *peer_id;
>>  };
>> @@ -133,6 +135,7 @@ struct pstream {
>>      const struct pstream_class *class;
>>      char *name;
>>      ovs_be16 bound_port;
>> +    replay_file_t replay_wfd;
>>  };
>>  
>>  void pstream_init(struct pstream *, const struct pstream_class *, char *name);
>> @@ -200,5 +203,7 @@ extern const struct pstream_class pwindows_pstream_class;
>>  extern const struct stream_class ssl_stream_class;
>>  extern const struct pstream_class pssl_pstream_class;
>>  #endif
>> +extern const struct stream_class replay_stream_class;
>> +extern const struct pstream_class preplay_pstream_class;
>>  
>>  #endif /* stream-provider.h */
>> diff --git a/lib/stream-replay.c b/lib/stream-replay.c
>> new file mode 100644
>> index 000000000..ef591b920
>> --- /dev/null
>> +++ b/lib/stream-replay.c
>> @@ -0,0 +1,459 @@
>> +/*
>> + * Copyright (c) 2021, Red Hat, Inc.
>> + *
>> + * Licensed under the Apache License, Version 2.0 (the "License");
>> + * you may not use this file except in compliance with the License.
>> + * You may obtain a copy of the License at:
>> + *
>> + *     http://www.apache.org/licenses/LICENSE-2.0
>> + *
>> + * Unless required by applicable law or agreed to in writing, software
>> + * distributed under the License is distributed on an "AS IS" BASIS,
>> + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>> + * See the License for the specific language governing permissions and
>> + * limitations under the License.
>> + */
>> +
>> +#include <config.h>
>> +#include <ctype.h>
>> +#include <errno.h>
>> +#include <poll.h>
>> +#include <stdlib.h>
>> +#include <string.h>
>> +#include <sys/socket.h>
>> +#include <sys/types.h>
>> +#include <unistd.h>
>> +#include "ovs-atomic.h"
>> +#include "ovs-replay.h"
>> +#include "util.h"
>> +#include "stream-provider.h"
>> +#include "stream.h"
>> +#include "openvswitch/poll-loop.h"
>> +#include "openvswitch/vlog.h"
>> +
>> +VLOG_DEFINE_THIS_MODULE(stream_replay);
>> +
>> +/* Active replay stream. */
>> +
>> +struct stream_replay
>> +{
> 
> Nit: Curly brace on the line above, please.
> 

Fixed.

>> +    struct stream stream;
>> +    replay_file_t f;
>> +    int seqno;
>> +};
>> +
>> +const struct stream_class replay_stream_class;
>> +
>> +/* Creates a new stream named 'name' that will emulate sending and receiving
>> + * data using replay file and stores a pointer to the stream in '*streamp'.
>> + *
>> + * Takes ownership of 'name'.
>> + *
>> + * Returns 0 if successful, otherwise a positive errno value. */
>> +static int
>> +new_replay_stream(char *name, struct stream **streamp)
>> +{
>> +    struct stream_replay *s;
>> +    int seqno = 0, error = 0, open_result;
>> +    replay_file_t f;
>> +
>> +    ovs_replay_lock();
>> +    error = ovs_replay_file_open(name, &f, &seqno);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to open stream.", name);
>> +        goto unlock;
> 
> I didn't try this out but don't we leak 'name' here?

Thanks!  I removed the "ownership" comments and made this function
to accept a const string just like the similar pstream function.

> 
>> +    }
>> +
>> +    error = ovs_replay_read(f, NULL, 0, &open_result, &seqno, true);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to read 'open' record.", name);
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> 
> Same here.
> 
>> +    }
>> +
>> +    if (open_result) {
>> +        error = -open_result;
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> 
> Here too.
> 
>> +    }
>> +
>> +    s = xmalloc(sizeof *s);
>> +    stream_init(&s->stream, &replay_stream_class, 0, name);

And added xstrdup() here.

>> +    s->f = f;
>> +    s->seqno = seqno;
>> +    *streamp = &s->stream;
>> +unlock:
>> +    ovs_replay_unlock();
>> +    return error;
>> +}
>> +
>> +static struct stream_replay *
>> +stream_replay_cast(struct stream *stream)
>> +{
>> +    stream_assert_class(stream, &replay_stream_class);
>> +    return CONTAINER_OF(stream, struct stream_replay, stream);
>> +}
>> +
>> +void
>> +stream_replay_open_wfd(struct stream *s, int open_result, const char *name)
>> +{
>> +    int state = ovs_replay_get_state();
>> +    int error = 0;
>> +    replay_file_t f;
>> +
>> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
>> +        return;
>> +    }
>> +
>> +    ovs_replay_lock();
>> +    error = ovs_replay_file_open(name, &f, NULL);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to open replay file for stream.", name);
>> +        ovs_replay_unlock();
>> +        return;
>> +    }
>> +    ovs_replay_unlock();
>> +
>> +    if (ovs_replay_write(f, NULL, -open_result, true)) {
>> +        VLOG_ERR("%s: failed to write 'open' failure: %d",
>> +                 s->name, open_result);
>> +    }
>> +    if (open_result) {
>> +        /* We recorded failure to open the stream. */
>> +        ovs_replay_file_close(f);
>> +    } else {
>> +        s->replay_wfd = f;
>> +    }
>> +}
>> +
>> +void
>> +stream_replay_write(struct stream *s, const void *buffer, int n, bool is_read)
>> +{
>> +    int state = ovs_replay_get_state();
>> +
>> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
>> +        return;
>> +    }
>> +
>> +    if (ovs_replay_write(s->replay_wfd, buffer, n, is_read)) {
>> +        VLOG_ERR("%s: failed to write buffer.", s->name);
> 
> ovs_replay_write() rate limits error logs, should we do that here too?
> AFAICT this applies to all VLOG_ERR() calls in this file.

Sure.  I added rate-limits to all replay errors here and in ovs-replay.

> 
>> +    }
>> +}
>> +
>> +void
>> +stream_replay_close_wfd(struct stream *s)
>> +{
>> +    if (s->replay_wfd) {
>> +        ovs_replay_file_close(s->replay_wfd);
>> +    }
>> +}
>> +
>> +static int
>> +stream_replay_open(const char *name, char *suffix OVS_UNUSED,
>> +                   struct stream **streamp, uint8_t dscp OVS_UNUSED)
>> +{
>> +    return new_replay_stream(xstrdup(name), streamp);
>> +}
>> +
>> +static void
>> +stream_replay_close(struct stream *stream)
>> +{
>> +    struct stream_replay *s = stream_replay_cast(stream);
>> +    ovs_replay_file_close(s->f);
>> +    free(s);
>> +}
>> +
>> +static ssize_t
>> +stream_replay_recv(struct stream *stream, void *buffer, size_t n)
>> +{
>> +    struct stream_replay *s = stream_replay_cast(stream);
>> +    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
>> +    int error, len;
>> +
>> +    ovs_replay_lock();
>> +    ovs_assert(norm_seqno >= ovs_replay_seqno());
>> +
>> +    if (norm_seqno != ovs_replay_seqno()
>> +        || !ovs_replay_seqno_is_read(s->seqno)) {
>> +        error = EAGAIN;
>> +        goto unlock;
>> +    }
>> +
>> +    error = ovs_replay_read(s->f, buffer, n, &len, &s->seqno, true);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to read from replay file.", stream->name);
>> +        goto unlock;
>> +    }
>> +
>> +unlock:
>> +    ovs_replay_unlock();
>> +    return error ? -error : len;
>> +}
>> +
>> +static ssize_t
>> +stream_replay_send(struct stream *stream OVS_UNUSED,
>> +                   const void *buffer OVS_UNUSED, size_t n)
>> +{
>> +    struct stream_replay *s = stream_replay_cast(stream);
>> +    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
>> +    int error, len;
>> +
>> +    ovs_replay_lock();
>> +    ovs_assert(norm_seqno >= ovs_replay_seqno());
>> +
>> +    if (norm_seqno != ovs_replay_seqno()
>> +        || ovs_replay_seqno_is_read(s->seqno)) {
>> +        error = EAGAIN;
>> +        goto unlock;
>> +    }
>> +
>> +    error = ovs_replay_read(s->f, NULL, 0, &len, &s->seqno, false);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to read from replay file.", stream->name);
>> +        goto unlock;
>> +    }
>> +    ovs_assert(len < 0 || len <= n);
>> +
>> +unlock:
>> +    ovs_replay_unlock();
>> +    return error ? -error : len;
>> +}
>> +
>> +static void
>> +stream_replay_wait(struct stream *stream, enum stream_wait_type wait)
>> +{
>> +    struct stream_replay *s = stream_replay_cast(stream);
>> +    switch (wait) {
>> +    case STREAM_CONNECT:
>> +        /* Connect does nothing and always available. */
>> +        poll_immediate_wake();
>> +        break;
>> +
>> +    case STREAM_SEND:
>> +        if (s->seqno != INT_MAX && !ovs_replay_seqno_is_read(s->seqno)) {
>> +            /* Stream waits for write. */
>> +            poll_immediate_wake();
>> +        }
>> +        break;
>> +
>> +    case STREAM_RECV:
>> +        if (s->seqno != INT_MAX && ovs_replay_seqno_is_read(s->seqno)) {
>> +            /* We still have something to read. */
>> +            poll_immediate_wake();
>> +        }
>> +        break;
>> +
>> +    default:
>> +        OVS_NOT_REACHED();
>> +    }
>> +}
>> +
>> +const struct stream_class replay_stream_class = {
>> +    "replay",                   /* name */
>> +    false,                      /* needs_probes */
>> +    stream_replay_open,         /* open */
>> +    stream_replay_close,        /* close */
>> +    NULL,                       /* connect */
>> +    stream_replay_recv,         /* recv */
>> +    stream_replay_send,         /* send */
>> +    NULL,                       /* run */
>> +    NULL,                       /* run_wait */
>> +    stream_replay_wait,         /* wait */
>> +};
>> +
>> +/* Passive replay stream. */
>> +
>> +struct replay_pstream
>> +{
>> +    struct pstream pstream;
>> +    replay_file_t f;
>> +    int seqno;
>> +};
>> +
>> +const struct pstream_class preplay_pstream_class;
>> +
>> +static struct replay_pstream *
>> +replay_pstream_cast(struct pstream *pstream)
>> +{
>> +    pstream_assert_class(pstream, &preplay_pstream_class);
>> +    return CONTAINER_OF(pstream, struct replay_pstream, pstream);
>> +}
>> +
>> +/* Creates a new pstream named 'name' that will accept new replay connections
>> + * reading them from the replay file and stores a pointer to the stream in
>> + * '*pstreamp'.
>> + *
>> + * Takes ownership of 'name'.
>> + *
>> + * Returns 0 if successful, otherwise a positive errno value. */
>> +static int
>> +pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
>> +                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
>> +{
>> +    int seqno = 0, error = 0, listen_result;
>> +    replay_file_t f;
>> +
>> +    ovs_replay_lock();
>> +    error = ovs_replay_file_open(name, &f, &seqno);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to open pstream.", name);> +        goto unlock;
> 
> 'name' is leaked here.
> 
>> +    }
>> +
>> +    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to read 'listen' record.", name);
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> 
> Same here.
> 
>> +    }
>> +
>> +    if (listen_result) {
>> +        error = -listen_result;
>> +        ovs_replay_file_close(f);
>> +        goto unlock;
> 
> Here too.
> 
>> +    }
>> +
>> +    struct replay_pstream *ps = xmalloc(sizeof *ps);
>> +    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));
> 
> I guess the xstrdup(name) is not needed, otherwise we don't really take
> ownership of 'name' and we actually leak it.
> 
>> +    ps->f = f;
>> +    ps->seqno = seqno;
>> +    *pstreamp = &ps->pstream;
>> +unlock:
>> +    ovs_replay_unlock();
>> +    return error;
>> +}
>> +
>> +void
>> +pstream_replay_open_wfd(struct pstream *ps, int listen_result,
>> +                        const char *name)
>> +{
>> +    int state = ovs_replay_get_state();
>> +    int error = 0;
>> +    replay_file_t f;
>> +
>> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
>> +        return;
>> +    }
>> +
>> +    ovs_replay_lock();
>> +    error = ovs_replay_file_open(name, &f, NULL);
>> +    if (error) {
>> +        VLOG_ERR("%s: failed to open replay file for pstream.", name);
>> +        ovs_replay_unlock();
>> +        return;
>> +    }
>> +    ovs_replay_unlock();
>> +
>> +    if (ovs_replay_write(f, NULL, -listen_result, true)) {
>> +        VLOG_ERR("%s: failed to write 'listen' result: %d",
>> +                 ps->name, listen_result);
>> +    }
>> +
>> +    if (listen_result) {
>> +        /* We recorded failure to open the stream. */
>> +        ovs_replay_file_close(f);
>> +    } else {
>> +        ps->replay_wfd = f;
>> +    }
>> +}
>> +
>> +
>> +void
>> +pstream_replay_write_accept(struct pstream *ps, const struct stream *s,
>> +                            int accept_result)
>> +{
>> +    int state = ovs_replay_get_state();
>> +    int len;
>> +
>> +    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
>> +        return;
>> +    }
>> +
>> +    if (!accept_result) {
>> +        len = strlen(s->name);
>> +        if (ovs_replay_write(ps->replay_wfd, s->name, len, true)) {
>> +            VLOG_ERR("%s: failed to write accept name: %s", ps->name, s->name);
>> +        }
>> +    } else if (ovs_replay_write(ps->replay_wfd, NULL, -accept_result, true)) {
>> +        VLOG_ERR("%s: failed to write 'accept' failure: %d",
>> +                 ps->name, accept_result);
>> +    }
>> +}
>> +
>> +void
>> +pstream_replay_close_wfd(struct pstream *ps)
>> +{
>> +    if (ps->replay_wfd) {
>> +        ovs_replay_file_close(ps->replay_wfd);
>> +    }
>> +}
>> +
>> +
> 
> Nit: I'd remove the second newline.

Done.

> 
> Thanks,
> Dumitru
>
diff mbox series

Patch

diff --git a/lib/automake.mk b/lib/automake.mk
index b558692c6..db9017591 100644
--- a/lib/automake.mk
+++ b/lib/automake.mk
@@ -312,6 +312,7 @@  lib_libopenvswitch_la_SOURCES = \
 	lib/stream-fd.c \
 	lib/stream-fd.h \
 	lib/stream-provider.h \
+	lib/stream-replay.c \
 	lib/stream-ssl.h \
 	lib/stream-tcp.c \
 	lib/stream.c \
diff --git a/lib/stream-provider.h b/lib/stream-provider.h
index 75f4f059b..44e3c6431 100644
--- a/lib/stream-provider.h
+++ b/lib/stream-provider.h
@@ -18,6 +18,7 @@ 
 #define STREAM_PROVIDER_H 1
 
 #include <sys/types.h>
+#include "ovs-replay.h"
 #include "stream.h"
 
 /* Active stream connection. */
@@ -29,6 +30,7 @@  struct stream {
     const struct stream_class *class;
     int state;
     int error;
+    replay_file_t replay_wfd;
     char *name;
     char *peer_id;
 };
@@ -133,6 +135,7 @@  struct pstream {
     const struct pstream_class *class;
     char *name;
     ovs_be16 bound_port;
+    replay_file_t replay_wfd;
 };
 
 void pstream_init(struct pstream *, const struct pstream_class *, char *name);
@@ -200,5 +203,7 @@  extern const struct pstream_class pwindows_pstream_class;
 extern const struct stream_class ssl_stream_class;
 extern const struct pstream_class pssl_pstream_class;
 #endif
+extern const struct stream_class replay_stream_class;
+extern const struct pstream_class preplay_pstream_class;
 
 #endif /* stream-provider.h */
diff --git a/lib/stream-replay.c b/lib/stream-replay.c
new file mode 100644
index 000000000..ef591b920
--- /dev/null
+++ b/lib/stream-replay.c
@@ -0,0 +1,459 @@ 
+/*
+ * Copyright (c) 2021, Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <config.h>
+#include <ctype.h>
+#include <errno.h>
+#include <poll.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+#include "ovs-atomic.h"
+#include "ovs-replay.h"
+#include "util.h"
+#include "stream-provider.h"
+#include "stream.h"
+#include "openvswitch/poll-loop.h"
+#include "openvswitch/vlog.h"
+
+VLOG_DEFINE_THIS_MODULE(stream_replay);
+
+/* Active replay stream. */
+
+struct stream_replay
+{
+    struct stream stream;
+    replay_file_t f;
+    int seqno;
+};
+
+const struct stream_class replay_stream_class;
+
+/* Creates a new stream named 'name' that will emulate sending and receiving
+ * data using replay file and stores a pointer to the stream in '*streamp'.
+ *
+ * Takes ownership of 'name'.
+ *
+ * Returns 0 if successful, otherwise a positive errno value. */
+static int
+new_replay_stream(char *name, struct stream **streamp)
+{
+    struct stream_replay *s;
+    int seqno = 0, error = 0, open_result;
+    replay_file_t f;
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, &seqno);
+    if (error) {
+        VLOG_ERR("%s: failed to open stream.", name);
+        goto unlock;
+    }
+
+    error = ovs_replay_read(f, NULL, 0, &open_result, &seqno, true);
+    if (error) {
+        VLOG_ERR("%s: failed to read 'open' record.", name);
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    if (open_result) {
+        error = -open_result;
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    s = xmalloc(sizeof *s);
+    stream_init(&s->stream, &replay_stream_class, 0, name);
+    s->f = f;
+    s->seqno = seqno;
+    *streamp = &s->stream;
+unlock:
+    ovs_replay_unlock();
+    return error;
+}
+
+static struct stream_replay *
+stream_replay_cast(struct stream *stream)
+{
+    stream_assert_class(stream, &replay_stream_class);
+    return CONTAINER_OF(stream, struct stream_replay, stream);
+}
+
+void
+stream_replay_open_wfd(struct stream *s, int open_result, const char *name)
+{
+    int state = ovs_replay_get_state();
+    int error = 0;
+    replay_file_t f;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, NULL);
+    if (error) {
+        VLOG_ERR("%s: failed to open replay file for stream.", name);
+        ovs_replay_unlock();
+        return;
+    }
+    ovs_replay_unlock();
+
+    if (ovs_replay_write(f, NULL, -open_result, true)) {
+        VLOG_ERR("%s: failed to write 'open' failure: %d",
+                 s->name, open_result);
+    }
+    if (open_result) {
+        /* We recorded failure to open the stream. */
+        ovs_replay_file_close(f);
+    } else {
+        s->replay_wfd = f;
+    }
+}
+
+void
+stream_replay_write(struct stream *s, const void *buffer, int n, bool is_read)
+{
+    int state = ovs_replay_get_state();
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    if (ovs_replay_write(s->replay_wfd, buffer, n, is_read)) {
+        VLOG_ERR("%s: failed to write buffer.", s->name);
+    }
+}
+
+void
+stream_replay_close_wfd(struct stream *s)
+{
+    if (s->replay_wfd) {
+        ovs_replay_file_close(s->replay_wfd);
+    }
+}
+
+static int
+stream_replay_open(const char *name, char *suffix OVS_UNUSED,
+                   struct stream **streamp, uint8_t dscp OVS_UNUSED)
+{
+    return new_replay_stream(xstrdup(name), streamp);
+}
+
+static void
+stream_replay_close(struct stream *stream)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    ovs_replay_file_close(s->f);
+    free(s);
+}
+
+static ssize_t
+stream_replay_recv(struct stream *stream, void *buffer, size_t n)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
+    int error, len;
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || !ovs_replay_seqno_is_read(s->seqno)) {
+        error = EAGAIN;
+        goto unlock;
+    }
+
+    error = ovs_replay_read(s->f, buffer, n, &len, &s->seqno, true);
+    if (error) {
+        VLOG_ERR("%s: failed to read from replay file.", stream->name);
+        goto unlock;
+    }
+
+unlock:
+    ovs_replay_unlock();
+    return error ? -error : len;
+}
+
+static ssize_t
+stream_replay_send(struct stream *stream OVS_UNUSED,
+                   const void *buffer OVS_UNUSED, size_t n)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
+    int error, len;
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || ovs_replay_seqno_is_read(s->seqno)) {
+        error = EAGAIN;
+        goto unlock;
+    }
+
+    error = ovs_replay_read(s->f, NULL, 0, &len, &s->seqno, false);
+    if (error) {
+        VLOG_ERR("%s: failed to read from replay file.", stream->name);
+        goto unlock;
+    }
+    ovs_assert(len < 0 || len <= n);
+
+unlock:
+    ovs_replay_unlock();
+    return error ? -error : len;
+}
+
+static void
+stream_replay_wait(struct stream *stream, enum stream_wait_type wait)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    switch (wait) {
+    case STREAM_CONNECT:
+        /* Connect does nothing and always available. */
+        poll_immediate_wake();
+        break;
+
+    case STREAM_SEND:
+        if (s->seqno != INT_MAX && !ovs_replay_seqno_is_read(s->seqno)) {
+            /* Stream waits for write. */
+            poll_immediate_wake();
+        }
+        break;
+
+    case STREAM_RECV:
+        if (s->seqno != INT_MAX && ovs_replay_seqno_is_read(s->seqno)) {
+            /* We still have something to read. */
+            poll_immediate_wake();
+        }
+        break;
+
+    default:
+        OVS_NOT_REACHED();
+    }
+}
+
+const struct stream_class replay_stream_class = {
+    "replay",                   /* name */
+    false,                      /* needs_probes */
+    stream_replay_open,         /* open */
+    stream_replay_close,        /* close */
+    NULL,                       /* connect */
+    stream_replay_recv,         /* recv */
+    stream_replay_send,         /* send */
+    NULL,                       /* run */
+    NULL,                       /* run_wait */
+    stream_replay_wait,         /* wait */
+};
+
+/* Passive replay stream. */
+
+struct replay_pstream
+{
+    struct pstream pstream;
+    replay_file_t f;
+    int seqno;
+};
+
+const struct pstream_class preplay_pstream_class;
+
+static struct replay_pstream *
+replay_pstream_cast(struct pstream *pstream)
+{
+    pstream_assert_class(pstream, &preplay_pstream_class);
+    return CONTAINER_OF(pstream, struct replay_pstream, pstream);
+}
+
+/* Creates a new pstream named 'name' that will accept new replay connections
+ * reading them from the replay file and stores a pointer to the stream in
+ * '*pstreamp'.
+ *
+ * Takes ownership of 'name'.
+ *
+ * Returns 0 if successful, otherwise a positive errno value. */
+static int
+pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
+                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
+{
+    int seqno = 0, error = 0, listen_result;
+    replay_file_t f;
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, &seqno);
+    if (error) {
+        VLOG_ERR("%s: failed to open pstream.", name);
+        goto unlock;
+    }
+
+    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
+    if (error) {
+        VLOG_ERR("%s: failed to read 'listen' record.", name);
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    if (listen_result) {
+        error = -listen_result;
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    struct replay_pstream *ps = xmalloc(sizeof *ps);
+    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));
+    ps->f = f;
+    ps->seqno = seqno;
+    *pstreamp = &ps->pstream;
+unlock:
+    ovs_replay_unlock();
+    return error;
+}
+
+void
+pstream_replay_open_wfd(struct pstream *ps, int listen_result,
+                        const char *name)
+{
+    int state = ovs_replay_get_state();
+    int error = 0;
+    replay_file_t f;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, NULL);
+    if (error) {
+        VLOG_ERR("%s: failed to open replay file for pstream.", name);
+        ovs_replay_unlock();
+        return;
+    }
+    ovs_replay_unlock();
+
+    if (ovs_replay_write(f, NULL, -listen_result, true)) {
+        VLOG_ERR("%s: failed to write 'listen' result: %d",
+                 ps->name, listen_result);
+    }
+
+    if (listen_result) {
+        /* We recorded failure to open the stream. */
+        ovs_replay_file_close(f);
+    } else {
+        ps->replay_wfd = f;
+    }
+}
+
+
+void
+pstream_replay_write_accept(struct pstream *ps, const struct stream *s,
+                            int accept_result)
+{
+    int state = ovs_replay_get_state();
+    int len;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    if (!accept_result) {
+        len = strlen(s->name);
+        if (ovs_replay_write(ps->replay_wfd, s->name, len, true)) {
+            VLOG_ERR("%s: failed to write accept name: %s", ps->name, s->name);
+        }
+    } else if (ovs_replay_write(ps->replay_wfd, NULL, -accept_result, true)) {
+        VLOG_ERR("%s: failed to write 'accept' failure: %d",
+                 ps->name, accept_result);
+    }
+}
+
+void
+pstream_replay_close_wfd(struct pstream *ps)
+{
+    if (ps->replay_wfd) {
+        ovs_replay_file_close(ps->replay_wfd);
+    }
+}
+
+
+static void
+pstream_replay_close(struct pstream *pstream)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+
+    ovs_replay_file_close(ps->f);
+    free(ps);
+}
+
+#define MAX_NAME_LEN 65536
+
+static int
+pstream_replay_accept(struct pstream *pstream, struct stream **new_streamp)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+    int norm_seqno = ovs_replay_normalized_seqno(ps->seqno);
+    int retval, len;
+    char name[MAX_NAME_LEN];
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || !ovs_replay_seqno_is_read(ps->seqno)) {
+        retval = EAGAIN;
+        ovs_replay_unlock();
+        goto exit;
+    }
+
+    retval = ovs_replay_read(ps->f, name, MAX_NAME_LEN - 1,
+                             &len, &ps->seqno, true);
+    if (retval) {
+        VLOG_ERR("%s: failed to read from replay file.", pstream->name);
+        ovs_replay_unlock();
+        goto exit;
+    }
+
+    ovs_replay_unlock();
+
+    if (len > 0) {
+        name[len] = 0;
+        retval = new_replay_stream(xstrdup(name), new_streamp);
+    } else {
+        retval = -len;
+    }
+exit:
+    return retval;
+}
+
+static void
+pstream_replay_wait(struct pstream *pstream)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+
+    if (ps->seqno != INT_MAX) {
+        /* Replay always has something to say. */
+        poll_immediate_wake();
+    }
+}
+
+const struct pstream_class preplay_pstream_class = {
+    "preplay",
+    false,
+    pstream_replay_listen,
+    pstream_replay_close,
+    pstream_replay_accept,
+    pstream_replay_wait,
+};
diff --git a/lib/stream.c b/lib/stream.c
index e246b3773..1e3c8a24e 100644
--- a/lib/stream.c
+++ b/lib/stream.c
@@ -33,6 +33,7 @@ 
 #include "openvswitch/ofp-print.h"
 #include "openvswitch/ofpbuf.h"
 #include "openvswitch/vlog.h"
+#include "ovs-replay.h"
 #include "ovs-thread.h"
 #include "packets.h"
 #include "openvswitch/poll-loop.h"
@@ -185,6 +186,9 @@  stream_lookup_class(const char *name, const struct stream_class **classp)
         if (strlen(class->name) == prefix_len
             && !memcmp(class->name, name, prefix_len)) {
             *classp = class;
+            if (ovs_replay_get_state() == OVS_REPLAY_READ) {
+                *classp = &replay_stream_class;
+            }
             return 0;
         }
     }
@@ -227,6 +231,8 @@  stream_open(const char *name, struct stream **streamp, uint8_t dscp)
     suffix_copy = xstrdup(strchr(name, ':') + 1);
     error = class->open(name, suffix_copy, &stream, dscp);
     free(suffix_copy);
+
+    stream_replay_open_wfd(stream, error, name);
     if (error) {
         goto error;
     }
@@ -295,6 +301,7 @@  stream_close(struct stream *stream)
     if (stream != NULL) {
         char *name = stream->name;
         char *peer_id = stream->peer_id;
+        stream_replay_close_wfd(stream);
         (stream->class->close)(stream);
         free(name);
         free(peer_id);
@@ -367,9 +374,13 @@  int
 stream_recv(struct stream *stream, void *buffer, size_t n)
 {
     int retval = stream_connect(stream);
-    return (retval ? -retval
-            : n == 0 ? 0
-            : (stream->class->recv)(stream, buffer, n));
+
+    retval = retval ? -retval
+             : n == 0 ? 0
+             : (stream->class->recv)(stream, buffer, n);
+
+    stream_replay_write(stream, buffer, retval, true);
+    return retval;
 }
 
 /* Tries to send up to 'n' bytes of 'buffer' on 'stream', and returns:
@@ -385,9 +396,12 @@  int
 stream_send(struct stream *stream, const void *buffer, size_t n)
 {
     int retval = stream_connect(stream);
-    return (retval ? -retval
-            : n == 0 ? 0
-            : (stream->class->send)(stream, buffer, n));
+    retval = retval ? -retval
+             : n == 0 ? 0
+             : (stream->class->send)(stream, buffer, n);
+
+    stream_replay_write(stream, buffer, retval, false);
+    return retval;
 }
 
 /* Allows 'stream' to perform maintenance activities, such as flushing
@@ -483,6 +497,9 @@  pstream_lookup_class(const char *name, const struct pstream_class **classp)
         if (strlen(class->name) == prefix_len
             && !memcmp(class->name, name, prefix_len)) {
             *classp = class;
+            if (ovs_replay_get_state() == OVS_REPLAY_READ) {
+                *classp = &preplay_pstream_class;
+            }
             return 0;
         }
     }
@@ -544,6 +561,8 @@  pstream_open(const char *name, struct pstream **pstreamp, uint8_t dscp)
     suffix_copy = xstrdup(strchr(name, ':') + 1);
     error = class->listen(name, suffix_copy, &pstream, dscp);
     free(suffix_copy);
+
+    pstream_replay_open_wfd(pstream, error, name);
     if (error) {
         goto error;
     }
@@ -571,6 +590,7 @@  pstream_close(struct pstream *pstream)
 {
     if (pstream != NULL) {
         char *name = pstream->name;
+        pstream_replay_close_wfd(pstream);
         (pstream->class->close)(pstream);
         free(name);
     }
@@ -588,9 +608,12 @@  pstream_accept(struct pstream *pstream, struct stream **new_stream)
     int retval = (pstream->class->accept)(pstream, new_stream);
     if (retval) {
         *new_stream = NULL;
+        pstream_replay_write_accept(pstream, NULL, retval);
     } else {
         ovs_assert((*new_stream)->state != SCS_CONNECTING
                    || (*new_stream)->class->connect);
+        pstream_replay_write_accept(pstream, *new_stream, 0);
+        stream_replay_open_wfd(*new_stream, 0, (*new_stream)->name);
     }
     return retval;
 }
diff --git a/lib/stream.h b/lib/stream.h
index 77bffa498..e30c51275 100644
--- a/lib/stream.h
+++ b/lib/stream.h
@@ -94,4 +94,16 @@  enum stream_content_type {
 void stream_report_content(const void *, ssize_t, enum stream_content_type,
                            struct vlog_module *, const char *stream_name);
 
+
+/* Stream replay helpers. */
+void stream_replay_open_wfd(struct stream *, int open_result,
+                            const char *name);
+void pstream_replay_open_wfd(struct pstream *, int listen_result,
+                             const char *name);
+void stream_replay_close_wfd(struct stream *);
+void pstream_replay_close_wfd(struct pstream *);
+void stream_replay_write(struct stream *, const void *, int, bool is_read);
+void pstream_replay_write_accept(struct pstream *, const struct stream *,
+                                 int accept_result);
+
 #endif /* stream.h */