lib/rconn.c

   1 /*
   2  * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18 #include "rconn.h"
  19 #include <errno.h>
  20 #include <limits.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include "coverage.h"
  24 #include "ofp-msgs.h"
  25 #include "ofp-util.h"
  26 #include "ofpbuf.h"
  27 #include "openflow/openflow.h"
  28 #include "poll-loop.h"
  29 #include "sat-math.h"
  30 #include "timeval.h"
  31 #include "util.h"
  32 #include "vconn.h"
  33 #include "vlog.h"
  34
  35 VLOG_DEFINE_THIS_MODULE(rconn);
  36
  37 COVERAGE_DEFINE(rconn_discarded);
  38 COVERAGE_DEFINE(rconn_overflow);
  39 COVERAGE_DEFINE(rconn_queued);
  40 COVERAGE_DEFINE(rconn_sent);
  41
  42 /* The connection states have the following meanings:
  43  *
  44  *    - S_VOID: No connection information is configured.
  45  *
  46  *    - S_BACKOFF: Waiting for a period of time before reconnecting.
  47  *
  48  *    - S_CONNECTING: A connection attempt is in progress and has not yet
  49  *      succeeded or failed.
  50  *
  51  *    - S_ACTIVE: A connection has been established and appears to be healthy.
  52  *
  53  *    - S_IDLE: A connection has been established but has been idle for some
  54  *      time.  An echo request has been sent, but no reply has yet been
  55  *      received.
  56  *
  57  *    - S_DISCONNECTED: An unreliable connection has disconnected and cannot be
  58  *      automatically retried.
  59  */
  60 #define STATES                                  \
  61     STATE(VOID, 1 << 0)                         \
  62     STATE(BACKOFF, 1 << 1)                      \
  63     STATE(CONNECTING, 1 << 2)                   \
  64     STATE(ACTIVE, 1 << 3)                       \
  65     STATE(IDLE, 1 << 4)                         \
  66     STATE(DISCONNECTED, 1 << 5)
  67 enum state {
  68 #define STATE(NAME, VALUE) S_##NAME = VALUE,
  69     STATES
  70 #undef STATE
  71 };
  72
  73 static const char *
  74 state_name(enum state state)
  75 {
  76     switch (state) {
  77 #define STATE(NAME, VALUE) case S_##NAME: return #NAME;
  78         STATES
  79 #undef STATE
  80     }
  81     return "***ERROR***";
  82 }
  83
  84 /* A reliable connection to an OpenFlow switch or controller.
  85  *
  86  * See the large comment in rconn.h for more information. */
  87 struct rconn {
  88     struct ovs_mutex mutex;
  89
  90     enum state state;
  91     time_t state_entered;
  92
  93     struct vconn *vconn;
  94     char *name;                 /* Human-readable descriptive name. */
  95     char *target;               /* vconn name, passed to vconn_open(). */
  96     bool reliable;
  97
  98     struct ovs_list txq;        /* Contains "struct ofpbuf"s. */
  99
 100     int backoff;
 101     int max_backoff;
 102     time_t backoff_deadline;
 103     time_t last_connected;
 104     time_t last_disconnected;
 105     unsigned int packets_sent;
 106     unsigned int seqno;
 107     int last_error;
 108
 109     /* In S_ACTIVE and S_IDLE, probably_admitted reports whether we believe
 110      * that the peer has made a (positive) admission control decision on our
 111      * connection.  If we have not yet been (probably) admitted, then the
 112      * connection does not reset the timer used for deciding whether the switch
 113      * should go into fail-open mode.
 114      *
 115      * last_admitted reports the last time we believe such a positive admission
 116      * control decision was made. */
 117     bool probably_admitted;
 118     time_t last_admitted;
 119
 120     /* These values are simply for statistics reporting, not used directly by
 121      * anything internal to the rconn (or ofproto for that matter). */
 122     unsigned int packets_received;
 123     unsigned int n_attempted_connections, n_successful_connections;
 124     time_t creation_time;
 125     unsigned long int total_time_connected;
 126
 127     /* Throughout this file, "probe" is shorthand for "inactivity probe".  When
 128      * no activity has been observed from the peer for a while, we send out an
 129      * echo request as an inactivity probe packet.  We should receive back a
 130      * response.
 131      *
 132      * "Activity" is defined as either receiving an OpenFlow message from the
 133      * peer or successfully sending a message that had been in 'txq'. */
 134     int probe_interval;         /* Secs of inactivity before sending probe. */
 135     time_t last_activity;       /* Last time we saw some activity. */
 136
 137     uint8_t dscp;
 138
 139     /* Messages sent or received are copied to the monitor connections. */
 140 #define MAXIMUM_MONITORS 8
 141     struct vconn *monitors[MAXIMUM_MONITORS];
 142     size_t n_monitors;
 143
 144     uint32_t allowed_versions;
 145 };
 146
 147 uint32_t rconn_get_allowed_versions(const struct rconn *rconn)
 148 {
 149     return rconn->allowed_versions;
 150 }
 151
 152 static unsigned int elapsed_in_this_state(const struct rconn *rc)
 153     OVS_REQUIRES(rc->mutex);
 154 static unsigned int timeout(const struct rconn *rc) OVS_REQUIRES(rc->mutex);
 155 static bool timed_out(const struct rconn *rc) OVS_REQUIRES(rc->mutex);
 156 static void state_transition(struct rconn *rc, enum state)
 157     OVS_REQUIRES(rc->mutex);
 158 static void rconn_set_target__(struct rconn *rc,
 159                                const char *target, const char *name)
 160     OVS_REQUIRES(rc->mutex);
 161 static int rconn_send__(struct rconn *rc, struct ofpbuf *,
 162                         struct rconn_packet_counter *)
 163     OVS_REQUIRES(rc->mutex);
 164 static int try_send(struct rconn *rc) OVS_REQUIRES(rc->mutex);
 165 static void reconnect(struct rconn *rc) OVS_REQUIRES(rc->mutex);
 166 static void report_error(struct rconn *rc, int error) OVS_REQUIRES(rc->mutex);
 167 static void rconn_disconnect__(struct rconn *rc) OVS_REQUIRES(rc->mutex);
 168 static void disconnect(struct rconn *rc, int error) OVS_REQUIRES(rc->mutex);
 169 static void flush_queue(struct rconn *rc) OVS_REQUIRES(rc->mutex);
 170 static void close_monitor(struct rconn *rc, size_t idx, int retval)
 171     OVS_REQUIRES(rc->mutex);
 172 static void copy_to_monitor(struct rconn *, const struct ofpbuf *);
 173 static bool is_connected_state(enum state);
 174 static bool is_admitted_msg(const struct ofpbuf *);
 175 static bool rconn_logging_connection_attempts__(const struct rconn *rc)
 176     OVS_REQUIRES(rc->mutex);
 177 static int rconn_get_version__(const struct rconn *rconn)
 178     OVS_REQUIRES(rconn->mutex);
 179
 180 /* The following prototypes duplicate those in rconn.h, but there we weren't
 181  * able to add the OVS_EXCLUDED annotations because the definition of struct
 182  * rconn was not visible. */
 183
 184 void rconn_set_max_backoff(struct rconn *rc, int max_backoff)
 185     OVS_EXCLUDED(rc->mutex);
 186 void rconn_connect(struct rconn *rc, const char *target, const char *name)
 187     OVS_EXCLUDED(rc->mutex);
 188 void rconn_connect_unreliably(struct rconn *rc,
 189                               struct vconn *vconn, const char *name)
 190     OVS_EXCLUDED(rc->mutex);
 191 void rconn_reconnect(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 192 void rconn_disconnect(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 193 void rconn_run(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 194 void rconn_run_wait(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 195 struct ofpbuf *rconn_recv(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 196 void rconn_recv_wait(struct rconn *rc) OVS_EXCLUDED(rc->mutex);
 197 int rconn_send(struct rconn *rc, struct ofpbuf *b,
 198                struct rconn_packet_counter *counter)
 199     OVS_EXCLUDED(rc->mutex);
 200 int rconn_send_with_limit(struct rconn *rc, struct ofpbuf *b,
 201                           struct rconn_packet_counter *counter,
 202                           int queue_limit)
 203     OVS_EXCLUDED(rc->mutex);
 204 void rconn_add_monitor(struct rconn *rc, struct vconn *vconn)
 205     OVS_EXCLUDED(rc->mutex);
 206 void rconn_set_name(struct rconn *rc, const char *new_name)
 207     OVS_EXCLUDED(rc->mutex);
 208 bool rconn_is_admitted(const struct rconn *rconn) OVS_EXCLUDED(rconn->mutex);
 209 int rconn_failure_duration(const struct rconn *rconn)
 210     OVS_EXCLUDED(rconn->mutex);
 211 ovs_be16 rconn_get_local_port(const struct rconn *rconn)
 212     OVS_EXCLUDED(rconn->mutex);
 213 int rconn_get_version(const struct rconn *rconn) OVS_EXCLUDED(rconn->mutex);
 214 unsigned int rconn_count_txqlen(const struct rconn *rc)
 215     OVS_EXCLUDED(rc->mutex);
 216
 217
 218 /* Creates and returns a new rconn.
 219  *
 220  * 'probe_interval' is a number of seconds.  If the interval passes once
 221  * without an OpenFlow message being received from the peer, the rconn sends
 222  * out an "echo request" message.  If the interval passes again without a
 223  * message being received, the rconn disconnects and re-connects to the peer.
 224  * Setting 'probe_interval' to 0 disables this behavior.
 225  *
 226  * 'max_backoff' is the maximum number of seconds between attempts to connect
 227  * to the peer.  The actual interval starts at 1 second and doubles on each
 228  * failure until it reaches 'max_backoff'.  If 0 is specified, the default of
 229  * 8 seconds is used.
 230  *
 231  * The new rconn is initially unconnected.  Use rconn_connect() or
 232  * rconn_connect_unreliably() to connect it.
 233  *
 234  * Connections made by the rconn will automatically negotiate an OpenFlow
 235  * protocol version acceptable to both peers on the connection.  The version
 236  * negotiated will be one of those in the 'allowed_versions' bitmap: version
 237  * 'x' is allowed if allowed_versions & (1 << x) is nonzero.  (The underlying
 238  * vconn will treat an 'allowed_versions' of 0 as OFPUTIL_DEFAULT_VERSIONS.)
 239  */
 240 struct rconn *
 241 rconn_create(int probe_interval, int max_backoff, uint8_t dscp,
 242              uint32_t allowed_versions)
 243 {
 244     struct rconn *rc = xzalloc(sizeof *rc);
 245
 246     ovs_mutex_init(&rc->mutex);
 247
 248     rc->state = S_VOID;
 249     rc->state_entered = time_now();
 250
 251     rc->vconn = NULL;
 252     rc->name = xstrdup("void");
 253     rc->target = xstrdup("void");
 254     rc->reliable = false;
 255
 256     list_init(&rc->txq);
 257
 258     rc->backoff = 0;
 259     rc->max_backoff = max_backoff ? max_backoff : 8;
 260     rc->backoff_deadline = TIME_MIN;
 261     rc->last_connected = TIME_MIN;
 262     rc->last_disconnected = TIME_MIN;
 263     rc->seqno = 0;
 264
 265     rc->packets_sent = 0;
 266
 267     rc->probably_admitted = false;
 268     rc->last_admitted = time_now();
 269
 270     rc->packets_received = 0;
 271     rc->n_attempted_connections = 0;
 272     rc->n_successful_connections = 0;
 273     rc->creation_time = time_now();
 274     rc->total_time_connected = 0;
 275
 276     rc->last_activity = time_now();
 277
 278     rconn_set_probe_interval(rc, probe_interval);
 279     rconn_set_dscp(rc, dscp);
 280
 281     rc->n_monitors = 0;
 282     rc->allowed_versions = allowed_versions;
 283
 284     return rc;
 285 }
 286
 287 void
 288 rconn_set_max_backoff(struct rconn *rc, int max_backoff)
 289     OVS_EXCLUDED(rc->mutex)
 290 {
 291     ovs_mutex_lock(&rc->mutex);
 292     rc->max_backoff = MAX(1, max_backoff);
 293     if (rc->state == S_BACKOFF && rc->backoff > max_backoff) {
 294         rc->backoff = max_backoff;
 295         if (rc->backoff_deadline > time_now() + max_backoff) {
 296             rc->backoff_deadline = time_now() + max_backoff;
 297         }
 298     }
 299     ovs_mutex_unlock(&rc->mutex);
 300 }
 301
 302 int
 303 rconn_get_max_backoff(const struct rconn *rc)
 304 {
 305     return rc->max_backoff;
 306 }
 307
 308 void
 309 rconn_set_dscp(struct rconn *rc, uint8_t dscp)
 310 {
 311     rc->dscp = dscp;
 312 }
 313
 314 uint8_t
 315 rconn_get_dscp(const struct rconn *rc)
 316 {
 317     return rc->dscp;
 318 }
 319
 320 void
 321 rconn_set_probe_interval(struct rconn *rc, int probe_interval)
 322 {
 323     rc->probe_interval = probe_interval ? MAX(5, probe_interval) : 0;
 324 }
 325
 326 int
 327 rconn_get_probe_interval(const struct rconn *rc)
 328 {
 329     return rc->probe_interval;
 330 }
 331
 332 /* Drops any existing connection on 'rc', then sets up 'rc' to connect to
 333  * 'target' and reconnect as needed.  'target' should be a remote OpenFlow
 334  * target in a form acceptable to vconn_open().
 335  *
 336  * If 'name' is nonnull, then it is used in log messages in place of 'target'.
 337  * It should presumably give more information to a human reader than 'target',
 338  * but it need not be acceptable to vconn_open(). */
 339 void
 340 rconn_connect(struct rconn *rc, const char *target, const char *name)
 341     OVS_EXCLUDED(rc->mutex)
 342 {
 343     ovs_mutex_lock(&rc->mutex);
 344     rconn_disconnect__(rc);
 345     rconn_set_target__(rc, target, name);
 346     rc->reliable = true;
 347     reconnect(rc);
 348     ovs_mutex_unlock(&rc->mutex);
 349 }
 350
 351 /* Drops any existing connection on 'rc', then configures 'rc' to use
 352  * 'vconn'.  If the connection on 'vconn' drops, 'rc' will not reconnect on it
 353  * own.
 354  *
 355  * By default, the target obtained from vconn_get_name(vconn) is used in log
 356  * messages.  If 'name' is nonnull, then it is used instead.  It should
 357  * presumably give more information to a human reader than the target, but it
 358  * need not be acceptable to vconn_open(). */
 359 void
 360 rconn_connect_unreliably(struct rconn *rc,
 361                          struct vconn *vconn, const char *name)
 362     OVS_EXCLUDED(rc->mutex)
 363 {
 364     ovs_assert(vconn != NULL);
 365
 366     ovs_mutex_lock(&rc->mutex);
 367     rconn_disconnect__(rc);
 368     rconn_set_target__(rc, vconn_get_name(vconn), name);
 369     rc->reliable = false;
 370     rc->vconn = vconn;
 371     rc->last_connected = time_now();
 372     state_transition(rc, S_ACTIVE);
 373     ovs_mutex_unlock(&rc->mutex);
 374 }
 375
 376 /* If 'rc' is connected, forces it to drop the connection and reconnect. */
 377 void
 378 rconn_reconnect(struct rconn *rc)
 379     OVS_EXCLUDED(rc->mutex)
 380 {
 381     ovs_mutex_lock(&rc->mutex);
 382     if (rc->state & (S_ACTIVE | S_IDLE)) {
 383         VLOG_INFO("%s: disconnecting", rc->name);
 384         disconnect(rc, 0);
 385     }
 386     ovs_mutex_unlock(&rc->mutex);
 387 }
 388
 389 static void
 390 rconn_disconnect__(struct rconn *rc)
 391     OVS_REQUIRES(rc->mutex)
 392 {
 393     if (rc->state != S_VOID) {
 394         if (rc->vconn) {
 395             vconn_close(rc->vconn);
 396             rc->vconn = NULL;
 397         }
 398         rconn_set_target__(rc, "void", NULL);
 399         rc->reliable = false;
 400
 401         rc->backoff = 0;
 402         rc->backoff_deadline = TIME_MIN;
 403
 404         state_transition(rc, S_VOID);
 405     }
 406 }
 407
 408 void
 409 rconn_disconnect(struct rconn *rc)
 410     OVS_EXCLUDED(rc->mutex)
 411 {
 412     ovs_mutex_lock(&rc->mutex);
 413     rconn_disconnect__(rc);
 414     ovs_mutex_unlock(&rc->mutex);
 415 }
 416
 417 /* Disconnects 'rc' and frees the underlying storage. */
 418 void
 419 rconn_destroy(struct rconn *rc)
 420 {
 421     if (rc) {
 422         size_t i;
 423
 424         ovs_mutex_lock(&rc->mutex);
 425         free(rc->name);
 426         free(rc->target);
 427         vconn_close(rc->vconn);
 428         flush_queue(rc);
 429         ofpbuf_list_delete(&rc->txq);
 430         for (i = 0; i < rc->n_monitors; i++) {
 431             vconn_close(rc->monitors[i]);
 432         }
 433         ovs_mutex_unlock(&rc->mutex);
 434         ovs_mutex_destroy(&rc->mutex);
 435
 436         free(rc);
 437     }
 438 }
 439
 440 static unsigned int
 441 timeout_VOID(const struct rconn *rc OVS_UNUSED)
 442     OVS_REQUIRES(rc->mutex)
 443 {
 444     return UINT_MAX;
 445 }
 446
 447 static void
 448 run_VOID(struct rconn *rc OVS_UNUSED)
 449     OVS_REQUIRES(rc->mutex)
 450 {
 451     /* Nothing to do. */
 452 }
 453
 454 static void
 455 reconnect(struct rconn *rc)
 456     OVS_REQUIRES(rc->mutex)
 457 {
 458     int retval;
 459
 460     if (rconn_logging_connection_attempts__(rc)) {
 461         VLOG_INFO("%s: connecting...", rc->name);
 462     }
 463     rc->n_attempted_connections++;
 464     retval = vconn_open(rc->target, rc->allowed_versions, rc->dscp,
 465                         &rc->vconn);
 466     if (!retval) {
 467         rc->backoff_deadline = time_now() + rc->backoff;
 468         state_transition(rc, S_CONNECTING);
 469     } else {
 470         VLOG_WARN("%s: connection failed (%s)",
 471                   rc->name, ovs_strerror(retval));
 472         rc->backoff_deadline = TIME_MAX; /* Prevent resetting backoff. */
 473         disconnect(rc, retval);
 474     }
 475 }
 476
 477 static unsigned int
 478 timeout_BACKOFF(const struct rconn *rc)
 479     OVS_REQUIRES(rc->mutex)
 480 {
 481     return rc->backoff;
 482 }
 483
 484 static void
 485 run_BACKOFF(struct rconn *rc)
 486     OVS_REQUIRES(rc->mutex)
 487 {
 488     if (timed_out(rc)) {
 489         reconnect(rc);
 490     }
 491 }
 492
 493 static unsigned int
 494 timeout_CONNECTING(const struct rconn *rc)
 495     OVS_REQUIRES(rc->mutex)
 496 {
 497     return MAX(1, rc->backoff);
 498 }
 499
 500 static void
 501 run_CONNECTING(struct rconn *rc)
 502     OVS_REQUIRES(rc->mutex)
 503 {
 504     int retval = vconn_connect(rc->vconn);
 505     if (!retval) {
 506         VLOG_INFO("%s: connected", rc->name);
 507         rc->n_successful_connections++;
 508         state_transition(rc, S_ACTIVE);
 509         rc->last_connected = rc->state_entered;
 510     } else if (retval != EAGAIN) {
 511         if (rconn_logging_connection_attempts__(rc)) {
 512             VLOG_INFO("%s: connection failed (%s)",
 513                       rc->name, ovs_strerror(retval));
 514         }
 515         disconnect(rc, retval);
 516     } else if (timed_out(rc)) {
 517         if (rconn_logging_connection_attempts__(rc)) {
 518             VLOG_INFO("%s: connection timed out", rc->name);
 519         }
 520         rc->backoff_deadline = TIME_MAX; /* Prevent resetting backoff. */
 521         disconnect(rc, ETIMEDOUT);
 522     }
 523 }
 524
 525 static void
 526 do_tx_work(struct rconn *rc)
 527     OVS_REQUIRES(rc->mutex)
 528 {
 529     if (list_is_empty(&rc->txq)) {
 530         return;
 531     }
 532     while (!list_is_empty(&rc->txq)) {
 533         int error = try_send(rc);
 534         if (error) {
 535             break;
 536         }
 537         rc->last_activity = time_now();
 538     }
 539     if (list_is_empty(&rc->txq)) {
 540         poll_immediate_wake();
 541     }
 542 }
 543
 544 static unsigned int
 545 timeout_ACTIVE(const struct rconn *rc)
 546     OVS_REQUIRES(rc->mutex)
 547 {
 548     if (rc->probe_interval) {
 549         unsigned int base = MAX(rc->last_activity, rc->state_entered);
 550         unsigned int arg = base + rc->probe_interval - rc->state_entered;
 551         return arg;
 552     }
 553     return UINT_MAX;
 554 }
 555
 556 static void
 557 run_ACTIVE(struct rconn *rc)
 558     OVS_REQUIRES(rc->mutex)
 559 {
 560     if (timed_out(rc)) {
 561         unsigned int base = MAX(rc->last_activity, rc->state_entered);
 562         int version;
 563
 564         VLOG_DBG("%s: idle %u seconds, sending inactivity probe",
 565                  rc->name, (unsigned int) (time_now() - base));
 566
 567         version = rconn_get_version__(rc);
 568         ovs_assert(version >= 0 && version <= 0xff);
 569
 570         /* Ordering is important here: rconn_send() can transition to BACKOFF,
 571          * and we don't want to transition back to IDLE if so, because then we
 572          * can end up queuing a packet with vconn == NULL and then *boom*. */
 573         state_transition(rc, S_IDLE);
 574         rconn_send__(rc, make_echo_request(version), NULL);
 575         return;
 576     }
 577
 578     do_tx_work(rc);
 579 }
 580
 581 static unsigned int
 582 timeout_IDLE(const struct rconn *rc)
 583     OVS_REQUIRES(rc->mutex)
 584 {
 585     return rc->probe_interval;
 586 }
 587
 588 static void
 589 run_IDLE(struct rconn *rc)
 590     OVS_REQUIRES(rc->mutex)
 591 {
 592     if (timed_out(rc)) {
 593         VLOG_ERR("%s: no response to inactivity probe after %u "
 594                  "seconds, disconnecting",
 595                  rc->name, elapsed_in_this_state(rc));
 596         disconnect(rc, ETIMEDOUT);
 597     } else {
 598         do_tx_work(rc);
 599     }
 600 }
 601
 602 static unsigned int
 603 timeout_DISCONNECTED(const struct rconn *rc OVS_UNUSED)
 604     OVS_REQUIRES(rc->mutex)
 605 {
 606     return UINT_MAX;
 607 }
 608
 609 static void
 610 run_DISCONNECTED(struct rconn *rc OVS_UNUSED)
 611     OVS_REQUIRES(rc->mutex)
 612 {
 613     /* Nothing to do. */
 614 }
 615
 616 /* Performs whatever activities are necessary to maintain 'rc': if 'rc' is
 617  * disconnected, attempts to (re)connect, backing off as necessary; if 'rc' is
 618  * connected, attempts to send packets in the send queue, if any. */
 619 void
 620 rconn_run(struct rconn *rc)
 621     OVS_EXCLUDED(rc->mutex)
 622 {
 623     int old_state;
 624     size_t i;
 625
 626     ovs_mutex_lock(&rc->mutex);
 627     if (rc->vconn) {
 628         int error;
 629
 630         vconn_run(rc->vconn);
 631
 632         error = vconn_get_status(rc->vconn);
 633         if (error) {
 634             report_error(rc, error);
 635             disconnect(rc, error);
 636         }
 637     }
 638     for (i = 0; i < rc->n_monitors; ) {
 639         struct ofpbuf *msg;
 640         int retval;
 641
 642         vconn_run(rc->monitors[i]);
 643
 644         /* Drain any stray message that came in on the monitor connection. */
 645         retval = vconn_recv(rc->monitors[i], &msg);
 646         if (!retval) {
 647             ofpbuf_delete(msg);
 648         } else if (retval != EAGAIN) {
 649             close_monitor(rc, i, retval);
 650             continue;
 651         }
 652         i++;
 653     }
 654
 655     do {
 656         old_state = rc->state;
 657         switch (rc->state) {
 658 #define STATE(NAME, VALUE) case S_##NAME: run_##NAME(rc); break;
 659             STATES
 660 #undef STATE
 661         default:
 662             OVS_NOT_REACHED();
 663         }
 664     } while (rc->state != old_state);
 665     ovs_mutex_unlock(&rc->mutex);
 666 }
 667
 668 /* Causes the next call to poll_block() to wake up when rconn_run() should be
 669  * called on 'rc'. */
 670 void
 671 rconn_run_wait(struct rconn *rc)
 672     OVS_EXCLUDED(rc->mutex)
 673 {
 674     unsigned int timeo;
 675     size_t i;
 676
 677     ovs_mutex_lock(&rc->mutex);
 678     if (rc->vconn) {
 679         vconn_run_wait(rc->vconn);
 680         if ((rc->state & (S_ACTIVE | S_IDLE)) && !list_is_empty(&rc->txq)) {
 681             vconn_wait(rc->vconn, WAIT_SEND);
 682         }
 683     }
 684     for (i = 0; i < rc->n_monitors; i++) {
 685         vconn_run_wait(rc->monitors[i]);
 686         vconn_recv_wait(rc->monitors[i]);
 687     }
 688
 689     timeo = timeout(rc);
 690     if (timeo != UINT_MAX) {
 691         long long int expires = sat_add(rc->state_entered, timeo);
 692         poll_timer_wait_until(expires * 1000);
 693     }
 694     ovs_mutex_unlock(&rc->mutex);
 695 }
 696
 697 /* Attempts to receive a packet from 'rc'.  If successful, returns the packet;
 698  * otherwise, returns a null pointer.  The caller is responsible for freeing
 699  * the packet (with ofpbuf_delete()). */
 700 struct ofpbuf *
 701 rconn_recv(struct rconn *rc)
 702     OVS_EXCLUDED(rc->mutex)
 703 {
 704     struct ofpbuf *buffer = NULL;
 705
 706     ovs_mutex_lock(&rc->mutex);
 707     if (rc->state & (S_ACTIVE | S_IDLE)) {
 708         int error = vconn_recv(rc->vconn, &buffer);
 709         if (!error) {
 710             copy_to_monitor(rc, buffer);
 711             if (rc->probably_admitted || is_admitted_msg(buffer)
 712                 || time_now() - rc->last_connected >= 30) {
 713                 rc->probably_admitted = true;
 714                 rc->last_admitted = time_now();
 715             }
 716             rc->last_activity = time_now();
 717             rc->packets_received++;
 718             if (rc->state == S_IDLE) {
 719                 state_transition(rc, S_ACTIVE);
 720             }
 721         } else if (error != EAGAIN) {
 722             report_error(rc, error);
 723             disconnect(rc, error);
 724         }
 725     }
 726     ovs_mutex_unlock(&rc->mutex);
 727
 728     return buffer;
 729 }
 730
 731 /* Causes the next call to poll_block() to wake up when a packet may be ready
 732  * to be received by vconn_recv() on 'rc'.  */
 733 void
 734 rconn_recv_wait(struct rconn *rc)
 735     OVS_EXCLUDED(rc->mutex)
 736 {
 737     ovs_mutex_lock(&rc->mutex);
 738     if (rc->vconn) {
 739         vconn_wait(rc->vconn, WAIT_RECV);
 740     }
 741     ovs_mutex_unlock(&rc->mutex);
 742 }
 743
 744 static int
 745 rconn_send__(struct rconn *rc, struct ofpbuf *b,
 746            struct rconn_packet_counter *counter)
 747     OVS_REQUIRES(rc->mutex)
 748 {
 749     if (rconn_is_connected(rc)) {
 750         COVERAGE_INC(rconn_queued);
 751         copy_to_monitor(rc, b);
 752
 753         if (counter) {
 754             rconn_packet_counter_inc(counter, ofpbuf_size(b));
 755         }
 756
 757         /* Reuse 'frame' as a private pointer while 'b' is in txq. */
 758         ofpbuf_set_frame(b, counter);
 759
 760         list_push_back(&rc->txq, &b->list_node);
 761
 762         /* If the queue was empty before we added 'b', try to send some
 763          * packets.  (But if the queue had packets in it, it's because the
 764          * vconn is backlogged and there's no point in stuffing more into it
 765          * now.  We'll get back to that in rconn_run().) */
 766         if (rc->txq.next == &b->list_node) {
 767             try_send(rc);
 768         }
 769         return 0;
 770     } else {
 771         ofpbuf_delete(b);
 772         return ENOTCONN;
 773     }
 774 }
 775
 776 /* Sends 'b' on 'rc'.  Returns 0 if successful, or ENOTCONN if 'rc' is not
 777  * currently connected.  Takes ownership of 'b'.
 778  *
 779  * If 'counter' is non-null, then 'counter' will be incremented while the
 780  * packet is in flight, then decremented when it has been sent (or discarded
 781  * due to disconnection).  Because 'b' may be sent (or discarded) before this
 782  * function returns, the caller may not be able to observe any change in
 783  * 'counter'.
 784  *
 785  * There is no rconn_send_wait() function: an rconn has a send queue that it
 786  * takes care of sending if you call rconn_run(), which will have the side
 787  * effect of waking up poll_block(). */
 788 int
 789 rconn_send(struct rconn *rc, struct ofpbuf *b,
 790            struct rconn_packet_counter *counter)
 791     OVS_EXCLUDED(rc->mutex)
 792 {
 793     int error;
 794
 795     ovs_mutex_lock(&rc->mutex);
 796     error = rconn_send__(rc, b, counter);
 797     ovs_mutex_unlock(&rc->mutex);
 798
 799     return error;
 800 }
 801
 802 /* Sends 'b' on 'rc'.  Increments 'counter' while the packet is in flight; it
 803  * will be decremented when it has been sent (or discarded due to
 804  * disconnection).  Returns 0 if successful, EAGAIN if 'counter->n' is already
 805  * at least as large as 'queue_limit', or ENOTCONN if 'rc' is not currently
 806  * connected.  Regardless of return value, 'b' is destroyed.
 807  *
 808  * Because 'b' may be sent (or discarded) before this function returns, the
 809  * caller may not be able to observe any change in 'counter'.
 810  *
 811  * There is no rconn_send_wait() function: an rconn has a send queue that it
 812  * takes care of sending if you call rconn_run(), which will have the side
 813  * effect of waking up poll_block(). */
 814 int
 815 rconn_send_with_limit(struct rconn *rc, struct ofpbuf *b,
 816                       struct rconn_packet_counter *counter, int queue_limit)
 817     OVS_EXCLUDED(rc->mutex)
 818 {
 819     int error;
 820
 821     ovs_mutex_lock(&rc->mutex);
 822     if (rconn_packet_counter_n_packets(counter) < queue_limit) {
 823         error = rconn_send__(rc, b, counter);
 824     } else {
 825         COVERAGE_INC(rconn_overflow);
 826         ofpbuf_delete(b);
 827         error = EAGAIN;
 828     }
 829     ovs_mutex_unlock(&rc->mutex);
 830
 831     return error;
 832 }
 833
 834 /* Returns the total number of packets successfully sent on the underlying
 835  * vconn.  A packet is not counted as sent while it is still queued in the
 836  * rconn, only when it has been successfuly passed to the vconn.  */
 837 unsigned int
 838 rconn_packets_sent(const struct rconn *rc)
 839 {
 840     return rc->packets_sent;
 841 }
 842
 843 /* Adds 'vconn' to 'rc' as a monitoring connection, to which all messages sent
 844  * and received on 'rconn' will be copied.  'rc' takes ownership of 'vconn'. */
 845 void
 846 rconn_add_monitor(struct rconn *rc, struct vconn *vconn)
 847     OVS_EXCLUDED(rc->mutex)
 848 {
 849     ovs_mutex_lock(&rc->mutex);
 850     if (rc->n_monitors < ARRAY_SIZE(rc->monitors)) {
 851         VLOG_INFO("new monitor connection from %s", vconn_get_name(vconn));
 852         rc->monitors[rc->n_monitors++] = vconn;
 853     } else {
 854         VLOG_DBG("too many monitor connections, discarding %s",
 855                  vconn_get_name(vconn));
 856         vconn_close(vconn);
 857     }
 858     ovs_mutex_unlock(&rc->mutex);
 859 }
 860
 861 /* Returns 'rc''s name.  This is a name for human consumption, appropriate for
 862  * use in log messages.  It is not necessarily a name that may be passed
 863  * directly to, e.g., vconn_open(). */
 864 const char *
 865 rconn_get_name(const struct rconn *rc)
 866 {
 867     return rc->name;
 868 }
 869
 870 /* Sets 'rc''s name to 'new_name'. */
 871 void
 872 rconn_set_name(struct rconn *rc, const char *new_name)
 873     OVS_EXCLUDED(rc->mutex)
 874 {
 875     ovs_mutex_lock(&rc->mutex);
 876     free(rc->name);
 877     rc->name = xstrdup(new_name);
 878     ovs_mutex_unlock(&rc->mutex);
 879 }
 880
 881 /* Returns 'rc''s target.  This is intended to be a string that may be passed
 882  * directly to, e.g., vconn_open(). */
 883 const char *
 884 rconn_get_target(const struct rconn *rc)
 885 {
 886     return rc->target;
 887 }
 888
 889 /* Returns true if 'rconn' is connected or in the process of reconnecting,
 890  * false if 'rconn' is disconnected and will not reconnect on its own. */
 891 bool
 892 rconn_is_alive(const struct rconn *rconn)
 893 {
 894     return rconn->state != S_VOID && rconn->state != S_DISCONNECTED;
 895 }
 896
 897 /* Returns true if 'rconn' is connected, false otherwise. */
 898 bool
 899 rconn_is_connected(const struct rconn *rconn)
 900 {
 901     return is_connected_state(rconn->state);
 902 }
 903
 904 static bool
 905 rconn_is_admitted__(const struct rconn *rconn)
 906     OVS_REQUIRES(rconn->mutex)
 907 {
 908     return (rconn_is_connected(rconn)
 909             && rconn->last_admitted >= rconn->last_connected);
 910 }
 911
 912 /* Returns true if 'rconn' is connected and thought to have been accepted by
 913  * the peer's admission-control policy. */
 914 bool
 915 rconn_is_admitted(const struct rconn *rconn)
 916     OVS_EXCLUDED(rconn->mutex)
 917 {
 918     bool admitted;
 919
 920     ovs_mutex_lock(&rconn->mutex);
 921     admitted = rconn_is_admitted__(rconn);
 922     ovs_mutex_unlock(&rconn->mutex);
 923
 924     return admitted;
 925 }
 926
 927 /* Returns 0 if 'rconn' is currently connected and considered to have been
 928  * accepted by the peer's admission-control policy, otherwise the number of
 929  * seconds since 'rconn' was last in such a state. */
 930 int
 931 rconn_failure_duration(const struct rconn *rconn)
 932     OVS_EXCLUDED(rconn->mutex)
 933 {
 934     int duration;
 935
 936     ovs_mutex_lock(&rconn->mutex);
 937     duration = (rconn_is_admitted__(rconn)
 938                 ? 0
 939                 : time_now() - rconn->last_admitted);
 940     ovs_mutex_unlock(&rconn->mutex);
 941
 942     return duration;
 943 }
 944
 945 static int
 946 rconn_get_version__(const struct rconn *rconn)
 947     OVS_REQUIRES(rconn->mutex)
 948 {
 949     return rconn->vconn ? vconn_get_version(rconn->vconn) : -1;
 950 }
 951
 952 /* Returns the OpenFlow version negotiated with the peer, or -1 if there is
 953  * currently no connection or if version negotiation is not yet complete. */
 954 int
 955 rconn_get_version(const struct rconn *rconn)
 956     OVS_EXCLUDED(rconn->mutex)
 957 {
 958     int version;
 959
 960     ovs_mutex_lock(&rconn->mutex);
 961     version = rconn_get_version__(rconn);
 962     ovs_mutex_unlock(&rconn->mutex);
 963
 964     return version;
 965 }
 966
 967 /* Returns the total number of packets successfully received by the underlying
 968  * vconn.  */
 969 unsigned int
 970 rconn_packets_received(const struct rconn *rc)
 971 {
 972     return rc->packets_received;
 973 }
 974
 975 /* Returns a string representing the internal state of 'rc'.  The caller must
 976  * not modify or free the string. */
 977 const char *
 978 rconn_get_state(const struct rconn *rc)
 979 {
 980     return state_name(rc->state);
 981 }
 982
 983 /* Returns the time at which the last successful connection was made by
 984  * 'rc'. Returns TIME_MIN if never connected. */
 985 time_t
 986 rconn_get_last_connection(const struct rconn *rc)
 987 {
 988     return rc->last_connected;
 989 }
 990
 991 /* Returns the time at which 'rc' was last disconnected. Returns TIME_MIN
 992  * if never disconnected. */
 993 time_t
 994 rconn_get_last_disconnect(const struct rconn *rc)
 995 {
 996     return rc->last_disconnected;
 997 }
 998
 999 /* Returns 'rc''s current connection sequence number, a number that changes
1000  * every time that 'rconn' connects or disconnects. */
1001 unsigned int
1002 rconn_get_connection_seqno(const struct rconn *rc)
1003 {
1004     return rc->seqno;
1005 }
1006
1007 /* Returns a value that explains why 'rc' last disconnected:
1008  *
1009  *   - 0 means that the last disconnection was caused by a call to
1010  *     rconn_disconnect(), or that 'rc' is new and has not yet completed its
1011  *     initial connection or connection attempt.
1012  *
1013  *   - EOF means that the connection was closed in the normal way by the peer.
1014  *
1015  *   - A positive integer is an errno value that represents the error.
1016  */
1017 int
1018 rconn_get_last_error(const struct rconn *rc)
1019 {
1020     return rc->last_error;
1021 }
1022
1023 /* Returns the number of messages queued for transmission on 'rc'. */
1024 unsigned int
1025 rconn_count_txqlen(const struct rconn *rc)
1026     OVS_EXCLUDED(rc->mutex)
1027 {
1028     unsigned int len;
1029
1030     ovs_mutex_lock(&rc->mutex);
1031     len = list_size(&rc->txq);
1032     ovs_mutex_unlock(&rc->mutex);
1033
1034     return len;
1035 }
1036 \f
1037 struct rconn_packet_counter *
1038 rconn_packet_counter_create(void)
1039 {
1040     struct rconn_packet_counter *c = xzalloc(sizeof *c);
1041     ovs_mutex_init(&c->mutex);
1042     ovs_mutex_lock(&c->mutex);
1043     c->ref_cnt = 1;
1044     ovs_mutex_unlock(&c->mutex);
1045     return c;
1046 }
1047
1048 void
1049 rconn_packet_counter_destroy(struct rconn_packet_counter *c)
1050 {
1051     if (c) {
1052         bool dead;
1053
1054         ovs_mutex_lock(&c->mutex);
1055         ovs_assert(c->ref_cnt > 0);
1056         dead = !--c->ref_cnt && !c->n_packets;
1057         ovs_mutex_unlock(&c->mutex);
1058
1059         if (dead) {
1060             ovs_mutex_destroy(&c->mutex);
1061             free(c);
1062         }
1063     }
1064 }
1065
1066 void
1067 rconn_packet_counter_inc(struct rconn_packet_counter *c, unsigned int n_bytes)
1068 {
1069     ovs_mutex_lock(&c->mutex);
1070     c->n_packets++;
1071     c->n_bytes += n_bytes;
1072     ovs_mutex_unlock(&c->mutex);
1073 }
1074
1075 void
1076 rconn_packet_counter_dec(struct rconn_packet_counter *c, unsigned int n_bytes)
1077 {
1078     bool dead = false;
1079
1080     ovs_mutex_lock(&c->mutex);
1081     ovs_assert(c->n_packets > 0);
1082     ovs_assert(c->n_packets == 1
1083                ? c->n_bytes == n_bytes
1084                : c->n_bytes > n_bytes);
1085     c->n_packets--;
1086     c->n_bytes -= n_bytes;
1087     dead = !c->n_packets && !c->ref_cnt;
1088     ovs_mutex_unlock(&c->mutex);
1089
1090     if (dead) {
1091         ovs_mutex_destroy(&c->mutex);
1092         free(c);
1093     }
1094 }
1095
1096 unsigned int
1097 rconn_packet_counter_n_packets(const struct rconn_packet_counter *c)
1098 {
1099     unsigned int n;
1100
1101     ovs_mutex_lock(&c->mutex);
1102     n = c->n_packets;
1103     ovs_mutex_unlock(&c->mutex);
1104
1105     return n;
1106 }
1107
1108 unsigned int
1109 rconn_packet_counter_n_bytes(const struct rconn_packet_counter *c)
1110 {
1111     unsigned int n;
1112
1113     ovs_mutex_lock(&c->mutex);
1114     n = c->n_bytes;
1115     ovs_mutex_unlock(&c->mutex);
1116
1117     return n;
1118 }
1119 \f
1120 /* Set rc->target and rc->name to 'target' and 'name', respectively.  If 'name'
1121  * is null, 'target' is used. */
1122 static void
1123 rconn_set_target__(struct rconn *rc, const char *target, const char *name)
1124     OVS_REQUIRES(rc->mutex)
1125 {
1126     free(rc->name);
1127     rc->name = xstrdup(name ? name : target);
1128     free(rc->target);
1129     rc->target = xstrdup(target);
1130 }
1131
1132 /* Tries to send a packet from 'rc''s send buffer.  Returns 0 if successful,
1133  * otherwise a positive errno value. */
1134 static int
1135 try_send(struct rconn *rc)
1136     OVS_REQUIRES(rc->mutex)
1137 {
1138     struct ofpbuf *msg = ofpbuf_from_list(rc->txq.next);
1139     unsigned int n_bytes = ofpbuf_size(msg);
1140     struct rconn_packet_counter *counter = msg->frame;
1141     int retval;
1142
1143     /* Eagerly remove 'msg' from the txq.  We can't remove it from the list
1144      * after sending, if sending is successful, because it is then owned by the
1145      * vconn, which might have freed it already. */
1146     list_remove(&msg->list_node);
1147     ofpbuf_set_frame(msg, NULL);
1148
1149     retval = vconn_send(rc->vconn, msg);
1150     if (retval) {
1151         ofpbuf_set_frame(msg, counter);
1152         list_push_front(&rc->txq, &msg->list_node);
1153         if (retval != EAGAIN) {
1154             report_error(rc, retval);
1155             disconnect(rc, retval);
1156         }
1157         return retval;
1158     }
1159     COVERAGE_INC(rconn_sent);
1160     rc->packets_sent++;
1161     if (counter) {
1162         rconn_packet_counter_dec(counter, n_bytes);
1163     }
1164     return 0;
1165 }
1166
1167 /* Reports that 'error' caused 'rc' to disconnect.  'error' may be a positive
1168  * errno value, or it may be EOF to indicate that the connection was closed
1169  * normally. */
1170 static void
1171 report_error(struct rconn *rc, int error)
1172     OVS_REQUIRES(rc->mutex)
1173 {
1174     /* On Windows, when a peer terminates without calling a closesocket()
1175      * on socket fd, we get WSAECONNRESET. Don't print warning messages
1176      * for that case. */
1177     if (error == EOF
1178 #ifdef _WIN32
1179         || error == WSAECONNRESET
1180 #endif
1181         ) {
1182         /* If 'rc' isn't reliable, then we don't really expect this connection
1183          * to last forever anyway (probably it's a connection that we received
1184          * via accept()), so use DBG level to avoid cluttering the logs. */
1185         enum vlog_level level = rc->reliable ? VLL_INFO : VLL_DBG;
1186         VLOG(level, "%s: connection closed by peer", rc->name);
1187     } else {
1188         VLOG_WARN("%s: connection dropped (%s)",
1189                   rc->name, ovs_strerror(error));
1190     }
1191 }
1192
1193 /* Disconnects 'rc' and records 'error' as the error that caused 'rc''s last
1194  * disconnection:
1195  *
1196  *   - 0 means that this disconnection is due to a request by 'rc''s client,
1197  *     not due to any kind of network error.
1198  *
1199  *   - EOF means that the connection was closed in the normal way by the peer.
1200  *
1201  *   - A positive integer is an errno value that represents the error.
1202  */
1203 static void
1204 disconnect(struct rconn *rc, int error)
1205     OVS_REQUIRES(rc->mutex)
1206 {
1207     rc->last_error = error;
1208     if (rc->vconn) {
1209         vconn_close(rc->vconn);
1210         rc->vconn = NULL;
1211     }
1212     if (rc->reliable) {
1213         time_t now = time_now();
1214
1215         if (rc->state & (S_CONNECTING | S_ACTIVE | S_IDLE)) {
1216             rc->last_disconnected = now;
1217             flush_queue(rc);
1218         }
1219
1220         if (now >= rc->backoff_deadline) {
1221             rc->backoff = 1;
1222         } else if (rc->backoff < rc->max_backoff / 2) {
1223             rc->backoff = MAX(1, 2 * rc->backoff);
1224             VLOG_INFO("%s: waiting %d seconds before reconnect",
1225                       rc->name, rc->backoff);
1226         } else {
1227             if (rconn_logging_connection_attempts__(rc)) {
1228                 VLOG_INFO("%s: continuing to retry connections in the "
1229                           "background but suppressing further logging",
1230                           rc->name);
1231             }
1232             rc->backoff = rc->max_backoff;
1233         }
1234         rc->backoff_deadline = now + rc->backoff;
1235         state_transition(rc, S_BACKOFF);
1236     } else {
1237         rc->last_disconnected = time_now();
1238         state_transition(rc, S_DISCONNECTED);
1239     }
1240 }
1241
1242 /* Drops all the packets from 'rc''s send queue and decrements their queue
1243  * counts. */
1244 static void
1245 flush_queue(struct rconn *rc)
1246     OVS_REQUIRES(rc->mutex)
1247 {
1248     if (list_is_empty(&rc->txq)) {
1249         return;
1250     }
1251     while (!list_is_empty(&rc->txq)) {
1252         struct ofpbuf *b = ofpbuf_from_list(list_pop_front(&rc->txq));
1253         struct rconn_packet_counter *counter = b->frame;
1254         if (counter) {
1255             rconn_packet_counter_dec(counter, ofpbuf_size(b));
1256         }
1257         COVERAGE_INC(rconn_discarded);
1258         ofpbuf_delete(b);
1259     }
1260     poll_immediate_wake();
1261 }
1262
1263 static unsigned int
1264 elapsed_in_this_state(const struct rconn *rc)
1265     OVS_REQUIRES(rc->mutex)
1266 {
1267     return time_now() - rc->state_entered;
1268 }
1269
1270 static unsigned int
1271 timeout(const struct rconn *rc)
1272     OVS_REQUIRES(rc->mutex)
1273 {
1274     switch (rc->state) {
1275 #define STATE(NAME, VALUE) case S_##NAME: return timeout_##NAME(rc);
1276         STATES
1277 #undef STATE
1278     default:
1279         OVS_NOT_REACHED();
1280     }
1281 }
1282
1283 static bool
1284 timed_out(const struct rconn *rc)
1285     OVS_REQUIRES(rc->mutex)
1286 {
1287     return time_now() >= sat_add(rc->state_entered, timeout(rc));
1288 }
1289
1290 static void
1291 state_transition(struct rconn *rc, enum state state)
1292     OVS_REQUIRES(rc->mutex)
1293 {
1294     rc->seqno += (rc->state == S_ACTIVE) != (state == S_ACTIVE);
1295     if (is_connected_state(state) && !is_connected_state(rc->state)) {
1296         rc->probably_admitted = false;
1297     }
1298     if (rconn_is_connected(rc)) {
1299         rc->total_time_connected += elapsed_in_this_state(rc);
1300     }
1301     VLOG_DBG("%s: entering %s", rc->name, state_name(state));
1302     rc->state = state;
1303     rc->state_entered = time_now();
1304 }
1305
1306 static void
1307 close_monitor(struct rconn *rc, size_t idx, int retval)
1308     OVS_REQUIRES(rc->mutex)
1309 {
1310     VLOG_DBG("%s: closing monitor connection to %s: %s",
1311              rconn_get_name(rc), vconn_get_name(rc->monitors[idx]),
1312              ovs_retval_to_string(retval));
1313     rc->monitors[idx] = rc->monitors[--rc->n_monitors];
1314 }
1315
1316 static void
1317 copy_to_monitor(struct rconn *rc, const struct ofpbuf *b)
1318     OVS_REQUIRES(rc->mutex)
1319 {
1320     struct ofpbuf *clone = NULL;
1321     int retval;
1322     size_t i;
1323
1324     for (i = 0; i < rc->n_monitors; ) {
1325         struct vconn *vconn = rc->monitors[i];
1326
1327         if (!clone) {
1328             clone = ofpbuf_clone(b);
1329         }
1330         retval = vconn_send(vconn, clone);
1331         if (!retval) {
1332             clone = NULL;
1333         } else if (retval != EAGAIN) {
1334             close_monitor(rc, i, retval);
1335             continue;
1336         }
1337         i++;
1338     }
1339     ofpbuf_delete(clone);
1340 }
1341
1342 static bool
1343 is_connected_state(enum state state)
1344 {
1345     return (state & (S_ACTIVE | S_IDLE)) != 0;
1346 }
1347
1348 /* When a switch initially connects to a controller, the controller may spend a
1349  * little time examining the switch, looking at, for example, its datapath ID,
1350  * before it decides whether it is willing to control that switch.  At that
1351  * point, it either disconnects or starts controlling the switch.
1352  *
1353  * This function returns a guess to its caller about whether 'b' is OpenFlow
1354  * message that indicates that the controller has decided to control the
1355  * switch.  It returns false if the message is one that a controller typically
1356  * uses to determine whether a switch is admissible, true if the message is one
1357  * that would typically be used only after the controller has admitted the
1358  * switch. */
1359 static bool
1360 is_admitted_msg(const struct ofpbuf *b)
1361 {
1362     enum ofptype type;
1363     enum ofperr error;
1364
1365     error = ofptype_decode(&type, ofpbuf_data(b));
1366     if (error) {
1367         return false;
1368     }
1369
1370     switch (type) {
1371     case OFPTYPE_HELLO:
1372     case OFPTYPE_ERROR:
1373     case OFPTYPE_ECHO_REQUEST:
1374     case OFPTYPE_ECHO_REPLY:
1375     case OFPTYPE_FEATURES_REQUEST:
1376     case OFPTYPE_FEATURES_REPLY:
1377     case OFPTYPE_GET_CONFIG_REQUEST:
1378     case OFPTYPE_GET_CONFIG_REPLY:
1379     case OFPTYPE_SET_CONFIG:
1380     case OFPTYPE_QUEUE_GET_CONFIG_REQUEST:
1381     case OFPTYPE_QUEUE_GET_CONFIG_REPLY:
1382     case OFPTYPE_GET_ASYNC_REQUEST:
1383     case OFPTYPE_GET_ASYNC_REPLY:
1384     case OFPTYPE_GROUP_STATS_REQUEST:
1385     case OFPTYPE_GROUP_STATS_REPLY:
1386     case OFPTYPE_GROUP_DESC_STATS_REQUEST:
1387     case OFPTYPE_GROUP_DESC_STATS_REPLY:
1388     case OFPTYPE_GROUP_FEATURES_STATS_REQUEST:
1389     case OFPTYPE_GROUP_FEATURES_STATS_REPLY:
1390     case OFPTYPE_TABLE_FEATURES_STATS_REQUEST:
1391     case OFPTYPE_TABLE_FEATURES_STATS_REPLY:
1392     case OFPTYPE_BUNDLE_CONTROL:
1393     case OFPTYPE_BUNDLE_ADD_MESSAGE:
1394         return false;
1395
1396     case OFPTYPE_PACKET_IN:
1397     case OFPTYPE_FLOW_REMOVED:
1398     case OFPTYPE_PORT_STATUS:
1399     case OFPTYPE_PACKET_OUT:
1400     case OFPTYPE_FLOW_MOD:
1401     case OFPTYPE_GROUP_MOD:
1402     case OFPTYPE_PORT_MOD:
1403     case OFPTYPE_TABLE_MOD:
1404     case OFPTYPE_METER_MOD:
1405     case OFPTYPE_BARRIER_REQUEST:
1406     case OFPTYPE_BARRIER_REPLY:
1407     case OFPTYPE_DESC_STATS_REQUEST:
1408     case OFPTYPE_DESC_STATS_REPLY:
1409     case OFPTYPE_FLOW_STATS_REQUEST:
1410     case OFPTYPE_FLOW_STATS_REPLY:
1411     case OFPTYPE_AGGREGATE_STATS_REQUEST:
1412     case OFPTYPE_AGGREGATE_STATS_REPLY:
1413     case OFPTYPE_TABLE_STATS_REQUEST:
1414     case OFPTYPE_TABLE_STATS_REPLY:
1415     case OFPTYPE_PORT_STATS_REQUEST:
1416     case OFPTYPE_PORT_STATS_REPLY:
1417     case OFPTYPE_QUEUE_STATS_REQUEST:
1418     case OFPTYPE_QUEUE_STATS_REPLY:
1419     case OFPTYPE_PORT_DESC_STATS_REQUEST:
1420     case OFPTYPE_PORT_DESC_STATS_REPLY:
1421     case OFPTYPE_METER_STATS_REQUEST:
1422     case OFPTYPE_METER_STATS_REPLY:
1423     case OFPTYPE_METER_CONFIG_STATS_REQUEST:
1424     case OFPTYPE_METER_CONFIG_STATS_REPLY:
1425     case OFPTYPE_METER_FEATURES_STATS_REQUEST:
1426     case OFPTYPE_METER_FEATURES_STATS_REPLY:
1427     case OFPTYPE_ROLE_REQUEST:
1428     case OFPTYPE_ROLE_REPLY:
1429     case OFPTYPE_ROLE_STATUS:
1430     case OFPTYPE_SET_FLOW_FORMAT:
1431     case OFPTYPE_FLOW_MOD_TABLE_ID:
1432     case OFPTYPE_SET_PACKET_IN_FORMAT:
1433     case OFPTYPE_FLOW_AGE:
1434     case OFPTYPE_SET_ASYNC_CONFIG:
1435     case OFPTYPE_SET_CONTROLLER_ID:
1436     case OFPTYPE_FLOW_MONITOR_STATS_REQUEST:
1437     case OFPTYPE_FLOW_MONITOR_STATS_REPLY:
1438     case OFPTYPE_FLOW_MONITOR_CANCEL:
1439     case OFPTYPE_FLOW_MONITOR_PAUSED:
1440     case OFPTYPE_FLOW_MONITOR_RESUMED:
1441     default:
1442         return true;
1443     }
1444 }
1445
1446 /* Returns true if 'rc' is currently logging information about connection
1447  * attempts, false if logging should be suppressed because 'rc' hasn't
1448  * successuflly connected in too long. */
1449 static bool
1450 rconn_logging_connection_attempts__(const struct rconn *rc)
1451     OVS_REQUIRES(rc->mutex)
1452 {
1453     return rc->backoff < rc->max_backoff;
1454 }