/*

 * 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);

static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 25);

/* 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'.

 *

 * Returns 0 if successful, otherwise a positive errno value. */

static int

new_replay_stream(const 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_RL(&rl, "%s: failed to open stream.", name);

        goto unlock;

    }

    error = ovs_replay_read(f, NULL, 0, &open_result, &seqno, true);

    if (error) {

        VLOG_ERR_RL(&rl, "%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, xstrdup(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_RL(&rl, "%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_RL(&rl, "%s: failed to write 'open' result: %d",

                    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_RL(&rl, "%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(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_RL(&rl, "%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_RL(&rl, "%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'.

 *

 * 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_RL(&rl, "%s: failed to open pstream.", name);

        goto unlock;

    }

    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);

    if (error) {

        VLOG_ERR_RL(&rl, "%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_RL(&rl, "%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_RL(&rl, "%s: failed to write 'listen' result: %d",

                    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_RL(&rl, "%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_RL(&rl, "%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_RL(&rl, "%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(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,

};