ofproto: Fix uninitialized field in ofputil_flow_update.
[sliver-openvswitch.git] / ofproto / connmgr.c
index a94f291..05e69c7 100644 (file)
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009, 2010, 2011, 2012 Nicira Networks.
+ * Copyright (c) 2009, 2010, 2011, 2012 Nicira, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,8 @@
 #include "fail-open.h"
 #include "in-band.h"
 #include "odp-util.h"
+#include "ofp-actions.h"
+#include "ofp-msgs.h"
 #include "ofp-util.h"
 #include "ofpbuf.h"
 #include "ofproto-provider.h"
@@ -33,6 +35,7 @@
 #include "pktbuf.h"
 #include "rconn.h"
 #include "shash.h"
+#include "simap.h"
 #include "stream.h"
 #include "timeval.h"
 #include "vconn.h"
@@ -86,6 +89,13 @@ struct ofconn {
      * that the message might be generated, a 0-bit disables it. */
     uint32_t master_async_config[OAM_N_TYPES]; /* master, other */
     uint32_t slave_async_config[OAM_N_TYPES];  /* slave */
+
+    /* Flow monitors. */
+    struct hmap monitors;       /* Contains "struct ofmonitor"s. */
+    struct list updates;        /* List of "struct ofpbuf"s. */
+    bool sent_abbrev_update;    /* Does 'updates' contain NXFME_ABBREV? */
+    struct rconn_packet_counter *monitor_counter;
+    uint64_t monitor_paused;
 };
 
 static struct ofconn *ofconn_create(struct connmgr *, struct rconn *,
@@ -122,11 +132,12 @@ struct ofservice {
     int rate_limit;             /* Max packet-in rate in packets per second. */
     int burst_limit;            /* Limit on accumulating packet credits. */
     bool enable_async_msgs;     /* Initially enable async messages? */
+    uint8_t dscp;               /* DSCP Value for controller connection */
 };
 
 static void ofservice_reconfigure(struct ofservice *,
                                   const struct ofproto_controller *);
-static int ofservice_create(struct connmgr *, const char *target);
+static int ofservice_create(struct connmgr *, const char *target, uint8_t dscp);
 static void ofservice_destroy(struct connmgr *, struct ofservice *);
 static struct ofservice *ofservice_lookup(struct connmgr *,
                                           const char *target);
@@ -159,6 +170,8 @@ struct connmgr {
 
 static void update_in_band_remotes(struct connmgr *);
 static void add_snooper(struct connmgr *, struct vconn *);
+static void ofmonitor_run(struct connmgr *);
+static void ofmonitor_wait(struct connmgr *);
 
 /* Creates and returns a new connection manager owned by 'ofproto'.  'name' is
  * a name for the ofproto suitable for using in log messages.
@@ -264,6 +277,7 @@ connmgr_run(struct connmgr *mgr,
     LIST_FOR_EACH_SAFE (ofconn, next_ofconn, node, &mgr->all_conns) {
         ofconn_run(ofconn, handle_openflow);
     }
+    ofmonitor_run(mgr);
 
     /* Fail-open maintenance.  Do this after processing the ofconns since
      * fail-open checks the status of the controller rconn. */
@@ -275,12 +289,13 @@ connmgr_run(struct connmgr *mgr,
         struct vconn *vconn;
         int retval;
 
-        retval = pvconn_accept(ofservice->pvconn, OFP_VERSION, &vconn);
+        retval = pvconn_accept(ofservice->pvconn, OFP10_VERSION, &vconn);
         if (!retval) {
             struct rconn *rconn;
             char *name;
 
-            rconn = rconn_create(ofservice->probe_interval, 0);
+            /* Passing default value for creation of the rconn */
+            rconn = rconn_create(ofservice->probe_interval, 0, ofservice->dscp);
             name = ofconn_make_name(mgr, vconn_get_name(vconn));
             rconn_connect_unreliably(rconn, vconn, name);
             free(name);
@@ -298,7 +313,7 @@ connmgr_run(struct connmgr *mgr,
         struct vconn *vconn;
         int retval;
 
-        retval = pvconn_accept(mgr->snoops[i], OFP_VERSION, &vconn);
+        retval = pvconn_accept(mgr->snoops[i], OFP10_VERSION, &vconn);
         if (!retval) {
             add_snooper(mgr, vconn);
         } else if (retval != EAGAIN) {
@@ -322,6 +337,7 @@ connmgr_wait(struct connmgr *mgr, bool handling_openflow)
     LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
         ofconn_wait(ofconn, handling_openflow);
     }
+    ofmonitor_wait(mgr);
     if (handling_openflow && mgr->in_band) {
         in_band_wait(mgr->in_band);
     }
@@ -336,6 +352,30 @@ connmgr_wait(struct connmgr *mgr, bool handling_openflow)
     }
 }
 
+/* Adds some memory usage statistics for 'mgr' into 'usage', for use with
+ * memory_report(). */
+void
+connmgr_get_memory_usage(const struct connmgr *mgr, struct simap *usage)
+{
+    const struct ofconn *ofconn;
+    unsigned int packets = 0;
+    unsigned int ofconns = 0;
+
+    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
+        int i;
+
+        ofconns++;
+
+        packets += rconn_count_txqlen(ofconn->rconn);
+        for (i = 0; i < N_SCHEDULERS; i++) {
+            packets += pinsched_count_txqlen(ofconn->schedulers[i]);
+        }
+        packets += pktbuf_count_packets(ofconn->pktbuf);
+    }
+    simap_increase(usage, "ofconns", ofconns);
+    simap_increase(usage, "packets", packets);
+}
+
 /* Returns the ofproto that owns 'ofconn''s connmgr. */
 struct ofproto *
 ofconn_get_ofproto(const struct ofconn *ofconn)
@@ -358,7 +398,7 @@ connmgr_retry(struct connmgr *mgr)
 \f
 /* OpenFlow configuration. */
 
-static void add_controller(struct connmgr *, const char *target);
+static void add_controller(struct connmgr *, const char *target, uint8_t dscp);
 static struct ofconn *find_controller_by_target(struct connmgr *,
                                                 const char *target);
 static void update_fail_open(struct connmgr *);
@@ -437,7 +477,7 @@ connmgr_free_controller_info(struct shash *info)
     SHASH_FOR_EACH (node, info) {
         struct ofproto_controller_info *cinfo = node->data;
         while (cinfo->pairs.n) {
-            free((char *) cinfo->pairs.values[--cinfo->pairs.n]);
+            free(CONST_CAST(char *, cinfo->pairs.values[--cinfo->pairs.n]));
         }
         free(cinfo);
     }
@@ -465,11 +505,15 @@ connmgr_set_controllers(struct connmgr *mgr,
 
         if (!vconn_verify_name(c->target)) {
             if (!find_controller_by_target(mgr, c->target)) {
-                add_controller(mgr, c->target);
+                VLOG_INFO("%s: added primary controller \"%s\"",
+                          mgr->name, c->target);
+                add_controller(mgr, c->target, c->dscp);
             }
         } else if (!pvconn_verify_name(c->target)) {
             if (!ofservice_lookup(mgr, c->target)) {
-                ofservice_create(mgr, c->target);
+                VLOG_INFO("%s: added service controller \"%s\"",
+                          mgr->name, c->target);
+                ofservice_create(mgr, c->target, c->dscp);
             }
         } else {
             VLOG_WARN_RL(&rl, "%s: unsupported controller \"%s\"",
@@ -483,10 +527,13 @@ connmgr_set_controllers(struct connmgr *mgr,
     /* Delete controllers that are no longer configured.
      * Update configuration of all now-existing controllers. */
     HMAP_FOR_EACH_SAFE (ofconn, next_ofconn, hmap_node, &mgr->controllers) {
+        const char *target = ofconn_get_target(ofconn);
         struct ofproto_controller *c;
 
-        c = shash_find_data(&new_controllers, ofconn_get_target(ofconn));
+        c = shash_find_data(&new_controllers, target);
         if (!c) {
+            VLOG_INFO("%s: removed primary controller \"%s\"",
+                      mgr->name, target);
             ofconn_destroy(ofconn);
         } else {
             ofconn_reconfigure(ofconn, c);
@@ -496,11 +543,13 @@ connmgr_set_controllers(struct connmgr *mgr,
     /* Delete services that are no longer configured.
      * Update configuration of all now-existing services. */
     HMAP_FOR_EACH_SAFE (ofservice, next_ofservice, node, &mgr->services) {
+        const char *target = pvconn_get_name(ofservice->pvconn);
         struct ofproto_controller *c;
 
-        c = shash_find_data(&new_controllers,
-                            pvconn_get_name(ofservice->pvconn));
+        c = shash_find_data(&new_controllers, target);
         if (!c) {
+            VLOG_INFO("%s: removed service controller \"%s\"",
+                      mgr->name, target);
             ofservice_destroy(mgr, ofservice);
         } else {
             ofservice_reconfigure(ofservice, c);
@@ -559,12 +608,12 @@ connmgr_has_snoops(const struct connmgr *mgr)
 /* Creates a new controller for 'target' in 'mgr'.  update_controller() needs
  * to be called later to finish the new ofconn's configuration. */
 static void
-add_controller(struct connmgr *mgr, const char *target)
+add_controller(struct connmgr *mgr, const char *target, uint8_t dscp)
 {
     char *name = ofconn_make_name(mgr, target);
     struct ofconn *ofconn;
 
-    ofconn = ofconn_create(mgr, rconn_create(5, 8), OFCONN_PRIMARY, true);
+    ofconn = ofconn_create(mgr, rconn_create(5, 8, dscp), OFCONN_PRIMARY, true);
     ofconn->pktbuf = pktbuf_create();
     rconn_connect(ofconn->rconn, target, name);
     hmap_insert(&mgr->controllers, &ofconn->hmap_node, hash_string(target, 0));
@@ -672,7 +721,7 @@ set_pvconns(struct pvconn ***pvconnsp, size_t *n_pvconnsp,
         struct pvconn *pvconn;
         int error;
 
-        error = pvconn_open(name, &pvconn);
+        error = pvconn_open(name, &pvconn, 0);
         if (!error) {
             pvconns[n_pvconns++] = pvconn;
         } else {
@@ -893,15 +942,14 @@ ofconn_send_error(const struct ofconn *ofconn,
         static struct vlog_rate_limit err_rl = VLOG_RATE_LIMIT_INIT(10, 10);
 
         if (!VLOG_DROP_INFO(&err_rl)) {
-            const struct ofputil_msg_type *type;
             const char *type_name;
             size_t request_len;
+            enum ofpraw raw;
 
             request_len = ntohs(request->length);
-            type_name = (!ofputil_decode_msg_type_partial(request,
-                                                          MIN(64, request_len),
-                                                          &type)
-                         ? ofputil_msg_type_name(type)
+            type_name = (!ofpraw_decode_partial(&raw, request,
+                                                MIN(64, request_len))
+                         ? ofpraw_get_name(raw)
                          : "invalid");
 
             VLOG_INFO("%s: sending %s error reply to %s message",
@@ -965,6 +1013,9 @@ ofconn_create(struct connmgr *mgr, struct rconn *rconn, enum ofconn_type type,
 
     list_init(&ofconn->opgroups);
 
+    hmap_init(&ofconn->monitors);
+    list_init(&ofconn->updates);
+
     ofconn_flush(ofconn);
 
     return ofconn;
@@ -975,6 +1026,7 @@ ofconn_create(struct connmgr *mgr, struct rconn *rconn, enum ofconn_type type,
 static void
 ofconn_flush(struct ofconn *ofconn)
 {
+    struct ofmonitor *monitor, *next_monitor;
     int i;
 
     ofconn->role = NX_ROLE_OTHER;
@@ -1043,6 +1095,14 @@ ofconn_flush(struct ofconn *ofconn)
         memset(ofconn->slave_async_config, 0,
                sizeof ofconn->slave_async_config);
     }
+
+    HMAP_FOR_EACH_SAFE (monitor, next_monitor, ofconn_node,
+                        &ofconn->monitors) {
+        ofmonitor_destroy(monitor);
+    }
+    rconn_packet_counter_destroy(ofconn->monitor_counter);
+    ofconn->monitor_counter = rconn_packet_counter_create();
+    ofpbuf_list_delete(&ofconn->updates); /* ...but it should be empty. */
 }
 
 static void
@@ -1059,6 +1119,7 @@ ofconn_destroy(struct ofconn *ofconn)
     rconn_packet_counter_destroy(ofconn->packet_in_counter);
     rconn_packet_counter_destroy(ofconn->reply_counter);
     pktbuf_destroy(ofconn->pktbuf);
+    rconn_packet_counter_destroy(ofconn->monitor_counter);
     free(ofconn);
 }
 
@@ -1078,6 +1139,12 @@ ofconn_reconfigure(struct ofconn *ofconn, const struct ofproto_controller *c)
     rconn_set_probe_interval(ofconn->rconn, probe_interval);
 
     ofconn_set_rate_limit(ofconn, c->rate_limit, c->burst_limit);
+
+    /* If dscp value changed reconnect. */
+    if (c->dscp != rconn_get_dscp(ofconn->rconn)) {
+        rconn_set_dscp(ofconn->rconn, c->dscp);
+        rconn_reconnect(ofconn->rconn);
+    }
 }
 
 /* Returns true if it makes sense for 'ofconn' to receive and process OpenFlow
@@ -1085,7 +1152,7 @@ ofconn_reconfigure(struct ofconn *ofconn, const struct ofproto_controller *c)
 static bool
 ofconn_may_recv(const struct ofconn *ofconn)
 {
-    int count = rconn_packet_counter_read (ofconn->reply_counter);
+    int count = ofconn->reply_counter->n_packets;
     return (!ofconn->blocked || ofconn->retry) && count < OFCONN_REPLY_MAX;
 }
 
@@ -1222,35 +1289,32 @@ static void
 ofconn_send(const struct ofconn *ofconn, struct ofpbuf *msg,
             struct rconn_packet_counter *counter)
 {
-    update_openflow_length(msg);
-    if (rconn_send(ofconn->rconn, msg, counter)) {
-        ofpbuf_delete(msg);
-    }
+    ofpmsg_update_length(msg);
+    rconn_send(ofconn->rconn, msg, counter);
 }
 \f
 /* Sending asynchronous messages. */
 
-static void schedule_packet_in(struct ofconn *, struct ofputil_packet_in,
-                               const struct flow *);
+static void schedule_packet_in(struct ofconn *, struct ofputil_packet_in);
 
 /* Sends an OFPT_PORT_STATUS message with 'opp' and 'reason' to appropriate
  * controllers managed by 'mgr'. */
 void
-connmgr_send_port_status(struct connmgr *mgr, const struct ofp_phy_port *opp,
-                         uint8_t reason)
+connmgr_send_port_status(struct connmgr *mgr,
+                         const struct ofputil_phy_port *pp, uint8_t reason)
 {
     /* XXX Should limit the number of queued port status change messages. */
+    struct ofputil_port_status ps;
     struct ofconn *ofconn;
 
+    ps.reason = reason;
+    ps.desc = *pp;
     LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
         if (ofconn_receives_async_msg(ofconn, OAM_PORT_STATUS, reason)) {
-            struct ofp_port_status *ops;
-            struct ofpbuf *b;
+            struct ofpbuf *msg;
 
-            ops = make_openflow_xid(sizeof *ops, OFPT_PORT_STATUS, 0, &b);
-            ops->reason = reason;
-            ops->desc = *opp;
-            ofconn_send(ofconn, b, NULL);
+            msg = ofputil_encode_port_status(&ps, ofconn->protocol);
+            ofconn_send(ofconn, msg, NULL);
         }
     }
 }
@@ -1279,18 +1343,19 @@ connmgr_send_flow_removed(struct connmgr *mgr,
 }
 
 /* Given 'pin', sends an OFPT_PACKET_IN message to each OpenFlow controller as
- * necessary according to their individual configurations. */
+ * necessary according to their individual configurations.
+ *
+ * The caller doesn't need to fill in pin->buffer_id or pin->total_len. */
 void
 connmgr_send_packet_in(struct connmgr *mgr,
-                       const struct ofputil_packet_in *pin,
-                       const struct flow *flow)
+                       const struct ofputil_packet_in *pin)
 {
     struct ofconn *ofconn;
 
     LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
         if (ofconn_receives_async_msg(ofconn, OAM_PACKET_IN, pin->reason)
             && ofconn->controller_id == pin->controller_id) {
-            schedule_packet_in(ofconn, *pin, flow);
+            schedule_packet_in(ofconn, *pin);
         }
     }
 }
@@ -1305,15 +1370,15 @@ do_send_packet_in(struct ofpbuf *ofp_packet_in, void *ofconn_)
                           ofconn->packet_in_counter, 100);
 }
 
-/* Takes 'pin', whose packet has the flow specified by 'flow', composes an
- * OpenFlow packet-in message from it, and passes it to 'ofconn''s packet
- * scheduler for sending. */
+/* Takes 'pin', composes an OpenFlow packet-in message from it, and passes it
+ * to 'ofconn''s packet scheduler for sending. */
 static void
-schedule_packet_in(struct ofconn *ofconn, struct ofputil_packet_in pin,
-                   const struct flow *flow)
+schedule_packet_in(struct ofconn *ofconn, struct ofputil_packet_in pin)
 {
     struct connmgr *mgr = ofconn->connmgr;
 
+    pin.total_len = pin.packet_len;
+
     /* Get OpenFlow buffer_id. */
     if (pin.reason == OFPR_ACTION) {
         pin.buffer_id = UINT32_MAX;
@@ -1323,7 +1388,7 @@ schedule_packet_in(struct ofconn *ofconn, struct ofputil_packet_in pin,
         pin.buffer_id = UINT32_MAX;
     } else {
         pin.buffer_id = pktbuf_save(ofconn->pktbuf, pin.packet, pin.packet_len,
-                                    flow->in_port);
+                                    pin.fmd.in_port);
     }
 
     /* Figure out how much of the packet to send. */
@@ -1331,7 +1396,7 @@ schedule_packet_in(struct ofconn *ofconn, struct ofputil_packet_in pin,
         pin.send_len = pin.packet_len;
     } else {
         /* Caller should have initialized 'send_len' to 'max_len' specified in
-         * struct ofp_action_output. */
+         * output action. */
     }
     if (pin.buffer_id != UINT32_MAX) {
         pin.send_len = MIN(pin.send_len, ofconn->miss_send_len);
@@ -1341,8 +1406,9 @@ schedule_packet_in(struct ofconn *ofconn, struct ofputil_packet_in pin,
      * immediately call into do_send_packet_in() or it might buffer it for a
      * while (until a later call to pinsched_run()). */
     pinsched_send(ofconn->schedulers[pin.reason == OFPR_NO_MATCH ? 0 : 1],
-                  flow->in_port,
-                  ofputil_encode_packet_in(&pin, ofconn->packet_in_format),
+                  pin.fmd.in_port,
+                  ofputil_encode_packet_in(&pin, ofconn->protocol,
+                                           ofconn->packet_in_format),
                   do_send_packet_in, ofconn);
 }
 \f
@@ -1438,29 +1504,6 @@ connmgr_is_any_controller_admitted(const struct connmgr *mgr)
     }
     return false;
 }
-
-/* Sends 'packet' to each controller connected to 'mgr'.  Takes ownership of
- * 'packet'. */
-void
-connmgr_broadcast(struct connmgr *mgr, struct ofpbuf *packet)
-{
-    struct ofconn *ofconn, *prev;
-
-    prev = NULL;
-    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
-        if (prev) {
-            ofconn_send_reply(ofconn, ofpbuf_clone(packet));
-        }
-        if (rconn_is_connected(ofconn->rconn)) {
-            prev = ofconn;
-        }
-    }
-    if (prev) {
-        ofconn_send_reply(prev, packet);
-    } else {
-        ofpbuf_delete(packet);
-    }
-}
 \f
 /* In-band configuration. */
 
@@ -1556,15 +1599,17 @@ connmgr_flushed(struct connmgr *mgr)
      * traffic until a controller has been defined and it tells us to do so. */
     if (!connmgr_has_controllers(mgr)
         && mgr->fail_mode == OFPROTO_FAIL_STANDALONE) {
-        union ofp_action action;
-        struct cls_rule rule;
+        struct ofpbuf ofpacts;
+        struct match match;
+
+        ofpbuf_init(&ofpacts, OFPACT_OUTPUT_SIZE);
+        ofpact_put_OUTPUT(&ofpacts)->port = OFPP_NORMAL;
+        ofpact_pad(&ofpacts);
+
+        match_init_catchall(&match);
+        ofproto_add_flow(mgr->ofproto, &match, 0, ofpacts.data, ofpacts.size);
 
-        memset(&action, 0, sizeof action);
-        action.type = htons(OFPAT_OUTPUT);
-        action.output.len = htons(sizeof action);
-        action.output.port = htons(OFPP_NORMAL);
-        cls_rule_init_catchall(&rule, 0);
-        ofproto_add_flow(mgr->ofproto, &rule, &action, 1);
+        ofpbuf_uninit(&ofpacts);
     }
 }
 \f
@@ -1574,13 +1619,13 @@ connmgr_flushed(struct connmgr *mgr)
  * ofservice_reconfigure() must be called to fully configure the new
  * ofservice. */
 static int
-ofservice_create(struct connmgr *mgr, const char *target)
+ofservice_create(struct connmgr *mgr, const char *target, uint8_t dscp)
 {
     struct ofservice *ofservice;
     struct pvconn *pvconn;
     int error;
 
-    error = pvconn_open(target, &pvconn);
+    error = pvconn_open(target, &pvconn, dscp);
     if (error) {
         return error;
     }
@@ -1608,6 +1653,7 @@ ofservice_reconfigure(struct ofservice *ofservice,
     ofservice->rate_limit = c->rate_limit;
     ofservice->burst_limit = c->burst_limit;
     ofservice->enable_async_msgs = c->enable_async_msgs;
+    ofservice->dscp = c->dscp;
 }
 
 /* Finds and returns the ofservice within 'mgr' that has the given
@@ -1625,3 +1671,242 @@ ofservice_lookup(struct connmgr *mgr, const char *target)
     }
     return NULL;
 }
+\f
+/* Flow monitors (NXST_FLOW_MONITOR). */
+
+/* A counter incremented when something significant happens to an OpenFlow
+ * rule.
+ *
+ *     - When a rule is added, its 'add_seqno' and 'modify_seqno' are set to
+ *       the current value (which is then incremented).
+ *
+ *     - When a rule is modified, its 'modify_seqno' is set to the current
+ *       value (which is then incremented).
+ *
+ * Thus, by comparing an old value of monitor_seqno against a rule's
+ * 'add_seqno', one can tell whether the rule was added before or after the old
+ * value was read, and similarly for 'modify_seqno'.
+ *
+ * 32 bits should normally be sufficient (and would be nice, to save space in
+ * each rule) but then we'd have to have some special cases for wraparound.
+ *
+ * We initialize monitor_seqno to 1 to allow 0 to be used as an invalid
+ * value. */
+static uint64_t monitor_seqno = 1;
+
+COVERAGE_DEFINE(ofmonitor_pause);
+COVERAGE_DEFINE(ofmonitor_resume);
+
+enum ofperr
+ofmonitor_create(const struct ofputil_flow_monitor_request *request,
+                 struct ofconn *ofconn, struct ofmonitor **monitorp)
+{
+    struct ofmonitor *m;
+
+    *monitorp = NULL;
+
+    m = ofmonitor_lookup(ofconn, request->id);
+    if (m) {
+        return OFPERR_NXBRC_FM_DUPLICATE_ID;
+    }
+
+    m = xmalloc(sizeof *m);
+    m->ofconn = ofconn;
+    hmap_insert(&ofconn->monitors, &m->ofconn_node, hash_int(request->id, 0));
+    m->id = request->id;
+    m->flags = request->flags;
+    m->out_port = request->out_port;
+    m->table_id = request->table_id;
+    minimatch_init(&m->match, &request->match);
+
+    *monitorp = m;
+    return 0;
+}
+
+struct ofmonitor *
+ofmonitor_lookup(struct ofconn *ofconn, uint32_t id)
+{
+    struct ofmonitor *m;
+
+    HMAP_FOR_EACH_IN_BUCKET (m, ofconn_node, hash_int(id, 0),
+                             &ofconn->monitors) {
+        if (m->id == id) {
+            return m;
+        }
+    }
+    return NULL;
+}
+
+void
+ofmonitor_destroy(struct ofmonitor *m)
+{
+    if (m) {
+        hmap_remove(&m->ofconn->monitors, &m->ofconn_node);
+        free(m);
+    }
+}
+
+void
+ofmonitor_report(struct connmgr *mgr, struct rule *rule,
+                 enum nx_flow_update_event event,
+                 enum ofp_flow_removed_reason reason,
+                 const struct ofconn *abbrev_ofconn, ovs_be32 abbrev_xid)
+{
+    enum nx_flow_monitor_flags update;
+    struct ofconn *ofconn;
+
+    switch (event) {
+    case NXFME_ADDED:
+        update = NXFMF_ADD;
+        rule->add_seqno = rule->modify_seqno = monitor_seqno++;
+        break;
+
+    case NXFME_DELETED:
+        update = NXFMF_DELETE;
+        break;
+
+    case NXFME_MODIFIED:
+        update = NXFMF_MODIFY;
+        rule->modify_seqno = monitor_seqno++;
+        break;
+
+    default:
+    case NXFME_ABBREV:
+        NOT_REACHED();
+    }
+
+    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
+        enum nx_flow_monitor_flags flags = 0;
+        struct ofmonitor *m;
+
+        if (ofconn->monitor_paused) {
+            /* Only send NXFME_DELETED notifications for flows that were added
+             * before we paused. */
+            if (event != NXFME_DELETED
+                || rule->add_seqno > ofconn->monitor_paused) {
+                continue;
+            }
+        }
+
+        HMAP_FOR_EACH (m, ofconn_node, &ofconn->monitors) {
+            if (m->flags & update
+                && (m->table_id == 0xff || m->table_id == rule->table_id)
+                && ofoperation_has_out_port(rule->pending, m->out_port)
+                && cls_rule_is_loose_match(&rule->cr, &m->match)) {
+                flags |= m->flags;
+            }
+        }
+
+        if (flags) {
+            if (list_is_empty(&ofconn->updates)) {
+                ofputil_start_flow_update(&ofconn->updates);
+                ofconn->sent_abbrev_update = false;
+            }
+
+            if (ofconn != abbrev_ofconn || ofconn->monitor_paused) {
+                struct ofputil_flow_update fu;
+                struct match match;
+
+                fu.event = event;
+                fu.reason = event == NXFME_DELETED ? reason : 0;
+                fu.idle_timeout = rule->idle_timeout;
+                fu.hard_timeout = rule->hard_timeout;
+                fu.table_id = rule->table_id;
+                fu.cookie = rule->flow_cookie;
+                minimatch_expand(&rule->cr.match, &match);
+                fu.match = &match;
+                fu.priority = rule->cr.priority;
+                if (flags & NXFMF_ACTIONS) {
+                    fu.ofpacts = rule->ofpacts;
+                    fu.ofpacts_len = rule->ofpacts_len;
+                } else {
+                    fu.ofpacts = NULL;
+                    fu.ofpacts_len = 0;
+                }
+                ofputil_append_flow_update(&fu, &ofconn->updates);
+            } else if (!ofconn->sent_abbrev_update) {
+                struct ofputil_flow_update fu;
+
+                fu.event = NXFME_ABBREV;
+                fu.xid = abbrev_xid;
+                ofputil_append_flow_update(&fu, &ofconn->updates);
+
+                ofconn->sent_abbrev_update = true;
+            }
+        }
+    }
+}
+
+void
+ofmonitor_flush(struct connmgr *mgr)
+{
+    struct ofconn *ofconn;
+
+    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
+        struct ofpbuf *msg, *next;
+
+        LIST_FOR_EACH_SAFE (msg, next, list_node, &ofconn->updates) {
+            list_remove(&msg->list_node);
+            ofconn_send(ofconn, msg, ofconn->monitor_counter);
+            if (!ofconn->monitor_paused
+                && ofconn->monitor_counter->n_bytes > 128 * 1024) {
+                struct ofpbuf *pause;
+
+                COVERAGE_INC(ofmonitor_pause);
+                ofconn->monitor_paused = monitor_seqno++;
+                pause = ofpraw_alloc_xid(OFPRAW_NXT_FLOW_MONITOR_PAUSED,
+                                         OFP10_VERSION, htonl(0), 0);
+                ofconn_send(ofconn, pause, ofconn->monitor_counter);
+            }
+        }
+    }
+}
+
+static void
+ofmonitor_resume(struct ofconn *ofconn)
+{
+    struct ofpbuf *resumed;
+    struct ofmonitor *m;
+    struct list rules;
+    struct list msgs;
+
+    list_init(&rules);
+    HMAP_FOR_EACH (m, ofconn_node, &ofconn->monitors) {
+        ofmonitor_collect_resume_rules(m, ofconn->monitor_paused, &rules);
+    }
+
+    list_init(&msgs);
+    ofmonitor_compose_refresh_updates(&rules, &msgs);
+
+    resumed = ofpraw_alloc_xid(OFPRAW_NXT_FLOW_MONITOR_RESUMED, OFP10_VERSION,
+                               htonl(0), 0);
+    list_push_back(&msgs, &resumed->list_node);
+    ofconn_send_replies(ofconn, &msgs);
+
+    ofconn->monitor_paused = 0;
+}
+
+static void
+ofmonitor_run(struct connmgr *mgr)
+{
+    struct ofconn *ofconn;
+
+    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
+        if (ofconn->monitor_paused && !ofconn->monitor_counter->n_packets) {
+            COVERAGE_INC(ofmonitor_resume);
+            ofmonitor_resume(ofconn);
+        }
+    }
+}
+
+static void
+ofmonitor_wait(struct connmgr *mgr)
+{
+    struct ofconn *ofconn;
+
+    LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
+        if (ofconn->monitor_paused && !ofconn->monitor_counter->n_packets) {
+            poll_immediate_wake();
+        }
+    }
+}