+ if (retval > 0) {
+ ofpbuf_pull(txbuf, retval);
+ if (!ofpbuf_size(txbuf)) {
+ list_remove(&txbuf->list_node);
+ ofpbuf_delete(txbuf);
+ }
+ } else if (retval != -EAGAIN) {
+ error = -retval;
+ }
+ }
+
+ if (!error) {
+ if (ofpbuf_size(&s->rxbuf) < 2) {
+ n = 2 - ofpbuf_size(&s->rxbuf);
+ } else {
+ uint16_t frame_len;
+
+ frame_len = ntohs(get_unaligned_be16(ofpbuf_data(&s->rxbuf)));
+ if (frame_len < ETH_HEADER_LEN) {
+ error = EPROTO;
+ n = 0;
+ } else {
+ n = (2 + frame_len) - ofpbuf_size(&s->rxbuf);
+ }
+ }
+ }
+ if (!error) {
+ int retval;
+
+ ofpbuf_prealloc_tailroom(&s->rxbuf, n);
+ retval = stream_recv(s->stream, ofpbuf_tail(&s->rxbuf), n);
+
+ if (retval > 0) {
+ ofpbuf_set_size(&s->rxbuf, ofpbuf_size(&s->rxbuf) + retval);
+ if (retval == n && ofpbuf_size(&s->rxbuf) > 2) {
+ ofpbuf_pull(&s->rxbuf, 2);
+ netdev_dummy_queue_packet(dev,
+ ofpbuf_clone(&s->rxbuf));
+ ofpbuf_clear(&s->rxbuf);
+ }
+ } else if (retval != -EAGAIN) {
+ error = (retval < 0 ? -retval
+ : ofpbuf_size(&s->rxbuf) ? EPROTO
+ : EOF);
+ }
+ }
+
+ return error;
+}
+
+static void
+dummy_packet_stream_close(struct dummy_packet_stream *s)
+{
+ stream_close(s->stream);
+ ofpbuf_uninit(&s->rxbuf);
+ ofpbuf_list_delete(&s->txq);
+}
+
+static void
+dummy_packet_conn_init(struct dummy_packet_conn *conn)
+{
+ memset(conn, 0, sizeof *conn);
+ conn->type = NONE;
+}
+
+static void
+dummy_packet_conn_get_config(struct dummy_packet_conn *conn, struct smap *args)
+{
+
+ switch (conn->type) {
+ case PASSIVE:
+ smap_add(args, "pstream", pstream_get_name(conn->u.pconn.pstream));
+ break;
+
+ case ACTIVE:
+ smap_add(args, "stream", stream_get_name(conn->u.rconn.rstream->stream));
+ break;
+
+ case NONE:
+ default:
+ break;
+ }
+}
+
+static void
+dummy_packet_conn_close(struct dummy_packet_conn *conn)
+{
+ int i;
+ struct dummy_packet_pconn *pconn = &conn->u.pconn;
+ struct dummy_packet_rconn *rconn = &conn->u.rconn;
+
+ switch (conn->type) {
+ case PASSIVE:
+ pstream_close(pconn->pstream);
+ for (i = 0; i < pconn->n_streams; i++) {
+ dummy_packet_stream_close(&pconn->streams[i]);
+ }
+ free(pconn->streams);
+ pconn->pstream = NULL;
+ pconn->streams = NULL;
+ break;
+
+ case ACTIVE:
+ dummy_packet_stream_close(rconn->rstream);
+ free(rconn->rstream);
+ rconn->rstream = NULL;
+ reconnect_destroy(rconn->reconnect);
+ rconn->reconnect = NULL;
+ break;
+
+ case NONE:
+ default:
+ break;
+ }
+
+ conn->type = NONE;
+ memset(conn, 0, sizeof *conn);
+}
+
+static void
+dummy_packet_conn_set_config(struct dummy_packet_conn *conn,
+ const struct smap *args)
+{
+ const char *pstream = smap_get(args, "pstream");
+ const char *stream = smap_get(args, "stream");
+
+ if (pstream && stream) {
+ VLOG_WARN("Open failed: both %s and %s are configured",
+ pstream, stream);
+ return;
+ }
+
+ switch (conn->type) {
+ case PASSIVE:
+ if (!strcmp(pstream_get_name(conn->u.pconn.pstream), pstream)) {
+ return;
+ }
+ dummy_packet_conn_close(conn);
+ break;
+ case ACTIVE:
+ if (!strcmp(stream_get_name(conn->u.rconn.rstream->stream), stream)) {
+ return;
+ }
+ dummy_packet_conn_close(conn);
+ break;
+ case NONE:
+ default:
+ break;
+ }
+
+ if (pstream) {
+ int error;
+
+ error = pstream_open(pstream, &conn->u.pconn.pstream, DSCP_DEFAULT);
+ if (error) {
+ VLOG_WARN("%s: open failed (%s)", pstream, ovs_strerror(error));
+ } else {
+ conn->type = PASSIVE;
+ }
+ }
+
+ if (stream) {
+ int error;
+ struct stream *active_stream;
+ struct reconnect *reconnect;;
+
+ reconnect = reconnect_create(time_msec());
+ reconnect_set_name(reconnect, stream);
+ reconnect_set_passive(reconnect, false, time_msec());
+ reconnect_enable(reconnect, time_msec());
+ reconnect_set_backoff(reconnect, 100, INT_MAX);
+ reconnect_set_probe_interval(reconnect, 0);
+ conn->u.rconn.reconnect = reconnect;
+ conn->type = ACTIVE;
+
+ error = stream_open(stream, &active_stream, DSCP_DEFAULT);
+ conn->u.rconn.rstream = dummy_packet_stream_create(active_stream);
+
+ switch (error) {
+ case 0:
+ reconnect_connected(reconnect, time_msec());
+ break;
+
+ case EAGAIN:
+ reconnect_connecting(reconnect, time_msec());
+ break;
+
+ default:
+ reconnect_connect_failed(reconnect, time_msec(), error);
+ stream_close(active_stream);
+ conn->u.rconn.rstream->stream = NULL;
+ break;
+ }
+ }
+}
+
+static void
+dummy_pconn_run(struct netdev_dummy *dev)
+ OVS_REQUIRES(dev->mutex)
+{
+ struct stream *new_stream;
+ struct dummy_packet_pconn *pconn = &dev->conn.u.pconn;
+ int error;
+ size_t i;
+
+ error = pstream_accept(pconn->pstream, &new_stream);
+ if (!error) {
+ struct dummy_packet_stream *s;
+
+ pconn->streams = xrealloc(pconn->streams,
+ ((pconn->n_streams + 1)
+ * sizeof *s));
+ s = &pconn->streams[pconn->n_streams++];
+ dummy_packet_stream_init(s, new_stream);
+ } else if (error != EAGAIN) {
+ VLOG_WARN("%s: accept failed (%s)",
+ pstream_get_name(pconn->pstream), ovs_strerror(error));
+ pstream_close(pconn->pstream);
+ pconn->pstream = NULL;
+ dev->conn.type = NONE;
+ }
+
+ for (i = 0; i < pconn->n_streams; i++) {
+ struct dummy_packet_stream *s = &pconn->streams[i];
+
+ error = dummy_packet_stream_run(dev, s);
+ if (error) {
+ VLOG_DBG("%s: closing connection (%s)",
+ stream_get_name(s->stream),
+ ovs_retval_to_string(error));
+ dummy_packet_stream_close(s);
+ pconn->streams[i] = pconn->streams[--pconn->n_streams];
+ }
+ }
+}
+
+static void
+dummy_rconn_run(struct netdev_dummy *dev)
+OVS_REQUIRES(dev->mutex)
+{
+ struct dummy_packet_rconn *rconn = &dev->conn.u.rconn;
+
+ switch (reconnect_run(rconn->reconnect, time_msec())) {
+ case RECONNECT_CONNECT:
+ {
+ int error;
+
+ if (rconn->rstream->stream) {
+ error = stream_connect(rconn->rstream->stream);
+ } else {
+ error = stream_open(reconnect_get_name(rconn->reconnect),
+ &rconn->rstream->stream, DSCP_DEFAULT);
+ }
+
+ switch (error) {
+ case 0:
+ reconnect_connected(rconn->reconnect, time_msec());
+ break;
+
+ case EAGAIN:
+ reconnect_connecting(rconn->reconnect, time_msec());
+ break;
+
+ default:
+ reconnect_connect_failed(rconn->reconnect, time_msec(), error);
+ stream_close(rconn->rstream->stream);
+ rconn->rstream->stream = NULL;
+ break;
+ }
+ }
+ break;
+
+ case RECONNECT_DISCONNECT:
+ case RECONNECT_PROBE:
+ default:
+ break;
+ }
+
+ if (reconnect_is_connected(rconn->reconnect)) {
+ int err;
+
+ err = dummy_packet_stream_run(dev, rconn->rstream);
+
+ if (err) {
+ reconnect_disconnected(rconn->reconnect, time_msec(), err);
+ stream_close(rconn->rstream->stream);
+ rconn->rstream->stream = NULL;
+ }
+ }
+}
+
+static void
+dummy_packet_conn_run(struct netdev_dummy *dev)
+ OVS_REQUIRES(dev->mutex)
+{
+ switch (dev->conn.type) {
+ case PASSIVE:
+ dummy_pconn_run(dev);
+ break;
+
+ case ACTIVE:
+ dummy_rconn_run(dev);
+ break;
+
+ case NONE:
+ default:
+ break;
+ }
+}
+
+static void
+dummy_packet_conn_wait(struct dummy_packet_conn *conn)
+{
+ int i;
+ switch (conn->type) {
+ case PASSIVE:
+ pstream_wait(conn->u.pconn.pstream);
+ for (i = 0; i < conn->u.pconn.n_streams; i++) {
+ struct dummy_packet_stream *s = &conn->u.pconn.streams[i];
+ dummy_packet_stream_wait(s);
+ }
+ break;
+ case ACTIVE:
+ if (reconnect_is_connected(conn->u.rconn.reconnect)) {
+ dummy_packet_stream_wait(conn->u.rconn.rstream);
+ }
+ break;
+
+ case NONE:
+ default:
+ break;
+ }
+}
+
+static void
+dummy_packet_conn_send(struct dummy_packet_conn *conn,
+ const void *buffer, size_t size)
+{
+ int i;
+
+ switch (conn->type) {
+ case PASSIVE:
+ for (i = 0; i < conn->u.pconn.n_streams; i++) {
+ struct dummy_packet_stream *s = &conn->u.pconn.streams[i];
+
+ dummy_packet_stream_send(s, buffer, size);
+ pstream_wait(conn->u.pconn.pstream);
+ }
+ break;
+
+ case ACTIVE:
+ if (reconnect_is_connected(conn->u.rconn.reconnect)) {
+ dummy_packet_stream_send(conn->u.rconn.rstream, buffer, size);
+ dummy_packet_stream_wait(conn->u.rconn.rstream);
+ }
+ break;
+
+ case NONE:
+ default:
+ break;
+ }