lib/learning-switch.c

   1 /* Copyright (c) 2008 The Board of Trustees of The Leland Stanford
   2  * Junior University
   3  *
   4  * We are making the OpenFlow specification and associated documentation
   5  * (Software) available for public use and benefit with the expectation
   6  * that others will use, modify and enhance the Software and contribute
   7  * those enhancements back to the community. However, since we would
   8  * like to make the Software available for broadest use, with as few
   9  * restrictions as possible permission is hereby granted, free of
  10  * charge, to any person obtaining a copy of this Software to deal in
  11  * the Software under the copyrights without restriction, including
  12  * without limitation the rights to use, copy, modify, merge, publish,
  13  * distribute, sublicense, and/or sell copies of the Software, and to
  14  * permit persons to whom the Software is furnished to do so, subject to
  15  * the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be
  18  * included in all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  21  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  22  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  23  * NONINFRINGEMENT.  IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  24  * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  25  * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  26  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  27  * SOFTWARE.
  28  *
  29  * The name and trademarks of copyright holder(s) may NOT be used in
  30  * advertising or publicity pertaining to the Software or any
  31  * derivatives without specific, written prior permission.
  32  */
  33
  34 #include <config.h>
  35 #include "learning-switch.h"
  36
  37 #include <errno.h>
  38 #include <inttypes.h>
  39 #include <netinet/in.h>
  40 #include <stdlib.h>
  41 #include <time.h>
  42
  43 #include "flow.h"
  44 #include "mac-learning.h"
  45 #include "ofpbuf.h"
  46 #include "ofp-print.h"
  47 #include "openflow/openflow.h"
  48 #include "poll-loop.h"
  49 #include "queue.h"
  50 #include "rconn.h"
  51 #include "stp.h"
  52 #include "timeval.h"
  53 #include "vconn.h"
  54 #include "xtoxll.h"
  55
  56 #define THIS_MODULE VLM_learning_switch
  57 #include "vlog.h"
  58
  59 enum port_state {
  60     P_DISABLED = 1 << 0,
  61     P_LISTENING = 1 << 1,
  62     P_LEARNING = 1 << 2,
  63     P_FORWARDING = 1 << 3,
  64     P_BLOCKING = 1 << 4
  65 };
  66
  67 struct lswitch {
  68     /* If nonnegative, the switch sets up flows that expire after the given
  69      * number of seconds (or never expire, if the value is OFP_FLOW_PERMANENT).
  70      * Otherwise, the switch processes every packet. */
  71     int max_idle;
  72
  73     unsigned long long int datapath_id;
  74     uint32_t capabilities;
  75     time_t last_features_request;
  76     struct mac_learning *ml;    /* NULL to act as hub instead of switch. */
  77
  78     /* Number of outgoing queued packets on the rconn. */
  79     int n_queued;
  80
  81     /* Spanning tree protocol implementation.
  82      *
  83      * We implement STP states by, whenever a port's STP state changes,
  84      * querying all the flows on the switch and then deleting any of them that
  85      * are inappropriate for a port's STP state. */
  86     long long int next_query;   /* Next time at which to query all flows. */
  87     long long int last_query;   /* Last time we sent a query. */
  88     long long int last_reply;   /* Last time we received a query reply. */
  89     unsigned int port_states[STP_MAX_PORTS];
  90     uint32_t query_xid;         /* XID used for query. */
  91     int n_flows, n_no_recv, n_no_send;
  92 };
  93
  94 /* The log messages here could actually be useful in debugging, so keep the
  95  * rate limit relatively high. */
  96 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(30, 300);
  97
  98 static void queue_tx(struct lswitch *, struct rconn *, struct ofpbuf *);
  99 static void send_features_request(struct lswitch *, struct rconn *);
 100 static void schedule_query(struct lswitch *, long long int delay);
 101 static bool may_learn(const struct lswitch *, uint16_t port_no);
 102 static bool may_recv(const struct lswitch *, uint16_t port_no,
 103                      bool any_actions);
 104 static bool may_send(const struct lswitch *, uint16_t port_no);
 105
 106 typedef void packet_handler_func(struct lswitch *, struct rconn *, void *);
 107 static packet_handler_func process_switch_features;
 108 static packet_handler_func process_packet_in;
 109 static packet_handler_func process_echo_request;
 110 static packet_handler_func process_port_status;
 111 static packet_handler_func process_phy_port;
 112 static packet_handler_func process_stats_reply;
 113
 114 /* Creates and returns a new learning switch.
 115  *
 116  * If 'learn_macs' is true, the new switch will learn the ports on which MAC
 117  * addresses appear.  Otherwise, the new switch will flood all packets.
 118  *
 119  * If 'max_idle' is nonnegative, the new switch will set up flows that expire
 120  * after the given number of seconds (or never expire, if 'max_idle' is
 121  * OFP_FLOW_PERMANENT).  Otherwise, the new switch will process every packet.
 122  *
 123  * 'rconn' is used to send out an OpenFlow features request. */
 124 struct lswitch *
 125 lswitch_create(struct rconn *rconn, bool learn_macs, int max_idle)
 126 {
 127     struct lswitch *sw;
 128     size_t i;
 129
 130     sw = xcalloc(1, sizeof *sw);
 131     sw->max_idle = max_idle;
 132     sw->datapath_id = 0;
 133     sw->last_features_request = time_now() - 1;
 134     sw->ml = learn_macs ? mac_learning_create() : NULL;
 135     sw->next_query = LLONG_MIN;
 136     sw->last_query = LLONG_MIN;
 137     sw->last_reply = LLONG_MIN;
 138     for (i = 0; i < STP_MAX_PORTS; i++) {
 139         sw->port_states[i] = P_DISABLED;
 140     }
 141     send_features_request(sw, rconn);
 142     return sw;
 143 }
 144
 145 /* Destroys 'sw'. */
 146 void
 147 lswitch_destroy(struct lswitch *sw)
 148 {
 149     if (sw) {
 150         mac_learning_destroy(sw->ml);
 151         free(sw);
 152     }
 153 }
 154
 155 /* Takes care of necessary 'sw' activity, except for receiving packets (which
 156  * the caller must do). */
 157 void
 158 lswitch_run(struct lswitch *sw, struct rconn *rconn)
 159 {
 160     long long int now = time_msec();
 161
 162     /* If we're waiting for more replies, keeping waiting for up to 10 s. */
 163     if (sw->last_reply != LLONG_MIN) {
 164         if (now - sw->last_reply > 10000) {
 165             VLOG_ERR_RL(&rl, "%012llx: No more flow stat replies last 10 s",
 166                         sw->datapath_id);
 167             sw->last_reply = LLONG_MIN;
 168             sw->last_query = LLONG_MIN;
 169             schedule_query(sw, 0);
 170         } else {
 171             return;
 172         }
 173     }
 174
 175     /* If we're waiting for any reply at all, keep waiting for up to 10 s. */
 176     if (sw->last_query != LLONG_MIN) {
 177         if (now - sw->last_query > 10000) {
 178             VLOG_ERR_RL(&rl, "%012llx: No flow stat replies in last 10 s",
 179                         sw->datapath_id);
 180             sw->last_query = LLONG_MIN;
 181             schedule_query(sw, 0);
 182         } else {
 183             return;
 184         }
 185     }
 186
 187     /* If it's time to send another query, do so. */
 188     if (sw->next_query != LLONG_MIN && now >= sw->next_query) {
 189         sw->next_query = LLONG_MIN;
 190         if (!rconn_is_connected(rconn)) {
 191             schedule_query(sw, 1000);
 192         } else {
 193             struct ofp_stats_request *osr;
 194             struct ofp_flow_stats_request *ofsr;
 195             struct ofpbuf *b;
 196             int error;
 197
 198             VLOG_DBG("%012llx: Sending flow stats request to implement STP",
 199                      sw->datapath_id);
 200
 201             sw->last_query = now;
 202             sw->query_xid = random_uint32();
 203             sw->n_flows = 0;
 204             sw->n_no_recv = 0;
 205             sw->n_no_send = 0;
 206             osr = make_openflow_xid(sizeof *osr + sizeof *ofsr,
 207                                     OFPT_STATS_REQUEST, sw->query_xid, &b);
 208             osr->type = htons(OFPST_FLOW);
 209             osr->flags = htons(0);
 210             ofsr = (struct ofp_flow_stats_request *) osr->body;
 211             ofsr->match.wildcards = htonl(OFPFW_ALL);
 212             ofsr->table_id = 0xff;
 213             ofsr->out_port = htons(OFPP_NONE);
 214
 215             error = rconn_send(rconn, b, NULL);
 216             if (error) {
 217                 VLOG_WARN_RL(&rl, "%012llx: sending flow stats request "
 218                              "failed: %s", sw->datapath_id, strerror(error));
 219                 ofpbuf_delete(b);
 220                 schedule_query(sw, 1000);
 221             }
 222         }
 223     }
 224 }
 225
 226 static void
 227 wait_timeout(long long int started)
 228 {
 229     long long int now = time_msec();
 230     long long int timeout = 10000 - (now - started);
 231     if (timeout <= 0) {
 232         poll_immediate_wake();
 233     } else {
 234         poll_timer_wait(timeout);
 235     }
 236 }
 237
 238 void
 239 lswitch_wait(struct lswitch *sw)
 240 {
 241     if (sw->last_reply != LLONG_MIN) {
 242         wait_timeout(sw->last_reply);
 243     } else if (sw->last_query != LLONG_MIN) {
 244         wait_timeout(sw->last_query);
 245     }
 246 }
 247
 248 /* Processes 'msg', which should be an OpenFlow received on 'rconn', according
 249  * to the learning switch state in 'sw'.  The most likely result of processing
 250  * is that flow-setup and packet-out OpenFlow messages will be sent out on
 251  * 'rconn'.  */
 252 void
 253 lswitch_process_packet(struct lswitch *sw, struct rconn *rconn,
 254                        const struct ofpbuf *msg)
 255 {
 256     struct processor {
 257         uint8_t type;
 258         size_t min_size;
 259         packet_handler_func *handler;
 260     };
 261     static const struct processor processors[] = {
 262         {
 263             OFPT_ECHO_REQUEST,
 264             sizeof(struct ofp_header),
 265             process_echo_request
 266         },
 267         {
 268             OFPT_FEATURES_REPLY,
 269             sizeof(struct ofp_switch_features),
 270             process_switch_features
 271         },
 272         {
 273             OFPT_PACKET_IN,
 274             offsetof(struct ofp_packet_in, data),
 275             process_packet_in
 276         },
 277         {
 278             OFPT_PORT_STATUS,
 279             sizeof(struct ofp_port_status),
 280             process_port_status
 281         },
 282         {
 283             OFPT_STATS_REPLY,
 284             offsetof(struct ofp_stats_reply, body),
 285             process_stats_reply
 286         },
 287         {
 288             OFPT_FLOW_EXPIRED,
 289             sizeof(struct ofp_flow_expired),
 290             NULL
 291         },
 292     };
 293     const size_t n_processors = ARRAY_SIZE(processors);
 294     const struct processor *p;
 295     struct ofp_header *oh;
 296
 297     oh = msg->data;
 298     if (sw->datapath_id == 0
 299         && oh->type != OFPT_ECHO_REQUEST
 300         && oh->type != OFPT_FEATURES_REPLY) {
 301         send_features_request(sw, rconn);
 302         return;
 303     }
 304
 305     for (p = processors; p < &processors[n_processors]; p++) {
 306         if (oh->type == p->type) {
 307             if (msg->size < p->min_size) {
 308                 VLOG_WARN_RL(&rl, "%012llx: %s: too short (%zu bytes) for "
 309                              "type %"PRIu8" (min %zu)", sw->datapath_id,
 310                              rconn_get_name(rconn), msg->size, oh->type,
 311                              p->min_size);
 312                 return;
 313             }
 314             if (p->handler) {
 315                 (p->handler)(sw, rconn, msg->data);
 316             }
 317             return;
 318         }
 319     }
 320     if (VLOG_IS_DBG_ENABLED()) {
 321         char *p = ofp_to_string(msg->data, msg->size, 2);
 322         VLOG_DBG_RL(&rl, "%012llx: OpenFlow packet ignored: %s",
 323                     sw->datapath_id, p);
 324         free(p);
 325     }
 326 }
 327 \f
 328 static void
 329 send_features_request(struct lswitch *sw, struct rconn *rconn)
 330 {
 331     time_t now = time_now();
 332     if (now >= sw->last_features_request + 1) {
 333         struct ofpbuf *b;
 334         struct ofp_switch_config *osc;
 335
 336         /* Send OFPT_FEATURES_REQUEST. */
 337         make_openflow(sizeof(struct ofp_header), OFPT_FEATURES_REQUEST, &b);
 338         queue_tx(sw, rconn, b);
 339
 340         /* Send OFPT_SET_CONFIG. */
 341         osc = make_openflow(sizeof *osc, OFPT_SET_CONFIG, &b);
 342         osc->flags = htons(OFPC_SEND_FLOW_EXP);
 343         osc->miss_send_len = htons(OFP_DEFAULT_MISS_SEND_LEN);
 344         queue_tx(sw, rconn, b);
 345
 346         sw->last_features_request = now;
 347     }
 348 }
 349
 350 static void
 351 queue_tx(struct lswitch *sw, struct rconn *rconn, struct ofpbuf *b)
 352 {
 353     int retval = rconn_send_with_limit(rconn, b, &sw->n_queued, 10);
 354     if (retval && retval != ENOTCONN) {
 355         if (retval == EAGAIN) {
 356             VLOG_WARN_RL(&rl, "%012llx: %s: tx queue overflow",
 357                          sw->datapath_id, rconn_get_name(rconn));
 358         } else {
 359             VLOG_WARN_RL(&rl, "%012llx: %s: send: %s",
 360                          sw->datapath_id, rconn_get_name(rconn),
 361                          strerror(retval));
 362         }
 363     }
 364 }
 365
 366 static void
 367 schedule_query(struct lswitch *sw, long long int delay)
 368 {
 369     long long int now = time_msec();
 370     if (sw->next_query == LLONG_MIN || sw->next_query > now + delay) {
 371         sw->next_query = now + delay;
 372     }
 373 }
 374
 375 static void
 376 process_switch_features(struct lswitch *sw, struct rconn *rconn, void *osf_)
 377 {
 378     struct ofp_switch_features *osf = osf_;
 379     size_t n_ports = ((ntohs(osf->header.length)
 380                        - offsetof(struct ofp_switch_features, ports))
 381                       / sizeof *osf->ports);
 382     size_t i;
 383
 384     sw->datapath_id = ntohll(osf->datapath_id);
 385     sw->capabilities = ntohl(osf->capabilities);
 386     for (i = 0; i < n_ports; i++) {
 387         process_phy_port(sw, rconn, &osf->ports[i]);
 388     }
 389     if (sw->capabilities & OFPC_STP) {
 390         schedule_query(sw, 1000);
 391     }
 392 }
 393
 394 static void
 395 process_packet_in(struct lswitch *sw, struct rconn *rconn, void *opi_)
 396 {
 397     struct ofp_packet_in *opi = opi_;
 398     uint16_t in_port = ntohs(opi->in_port);
 399     uint16_t out_port = OFPP_FLOOD;
 400
 401     size_t pkt_ofs, pkt_len;
 402     struct ofpbuf pkt;
 403     struct flow flow;
 404
 405     /* Extract flow data from 'opi' into 'flow'. */
 406     pkt_ofs = offsetof(struct ofp_packet_in, data);
 407     pkt_len = ntohs(opi->header.length) - pkt_ofs;
 408     pkt.data = opi->data;
 409     pkt.size = pkt_len;
 410     flow_extract(&pkt, in_port, &flow);
 411
 412     if (may_learn(sw, in_port) && sw->ml) {
 413         if (mac_learning_learn(sw->ml, flow.dl_src, in_port)) {
 414             VLOG_DBG_RL(&rl, "%012llx: learned that "ETH_ADDR_FMT" is on "
 415                         "port %"PRIu16, sw->datapath_id,
 416                         ETH_ADDR_ARGS(flow.dl_src), in_port);
 417         }
 418     }
 419
 420     if (eth_addr_is_reserved(flow.dl_src)) {
 421         goto drop_it;
 422     }
 423
 424     if (!may_recv(sw, in_port, false)) {
 425         /* STP prevents receiving anything on this port. */
 426         goto drop_it;
 427     }
 428
 429     if (sw->ml) {
 430         uint16_t learned_port = mac_learning_lookup(sw->ml, flow.dl_dst);
 431         if (may_send(sw, learned_port)) {
 432             out_port = learned_port;
 433         }
 434     }
 435
 436     if (in_port == out_port) {
 437         /* Don't send out packets on their input ports. */
 438         goto drop_it;
 439     } else if (sw->max_idle >= 0 && (!sw->ml || out_port != OFPP_FLOOD)) {
 440         /* The output port is known, or we always flood everything, so add a
 441          * new flow. */
 442         queue_tx(sw, rconn, make_add_simple_flow(&flow, ntohl(opi->buffer_id),
 443                                                  out_port, sw->max_idle));
 444
 445         /* If the switch didn't buffer the packet, we need to send a copy. */
 446         if (ntohl(opi->buffer_id) == UINT32_MAX) {
 447             queue_tx(sw, rconn,
 448                      make_unbuffered_packet_out(&pkt, in_port, out_port));
 449         }
 450     } else {
 451         /* We don't know that MAC, or we don't set up flows.  Send along the
 452          * packet without setting up a flow. */
 453         struct ofpbuf *b;
 454         if (ntohl(opi->buffer_id) == UINT32_MAX) {
 455             b = make_unbuffered_packet_out(&pkt, in_port, out_port);
 456         } else {
 457             b = make_buffered_packet_out(ntohl(opi->buffer_id),
 458                                          in_port, out_port);
 459         }
 460         queue_tx(sw, rconn, b);
 461     }
 462     return;
 463
 464 drop_it:
 465     if (sw->max_idle >= 0) {
 466         /* Set up a flow to drop packets. */
 467         queue_tx(sw, rconn, make_add_flow(&flow, ntohl(opi->buffer_id),
 468                                           sw->max_idle, 0));
 469     } else {
 470         /* Just drop the packet, since we don't set up flows at all.
 471          * XXX we should send a packet_out with no actions if buffer_id !=
 472          * UINT32_MAX, to avoid clogging the kernel buffers. */
 473     }
 474     return;
 475 }
 476
 477 static void
 478 process_echo_request(struct lswitch *sw, struct rconn *rconn, void *rq_)
 479 {
 480     struct ofp_header *rq = rq_;
 481     queue_tx(sw, rconn, make_echo_reply(rq));
 482 }
 483
 484 static void
 485 process_port_status(struct lswitch *sw, struct rconn *rconn, void *ops_)
 486 {
 487     struct ofp_port_status *ops = ops_;
 488     process_phy_port(sw, rconn, &ops->desc);
 489 }
 490
 491 static void
 492 process_phy_port(struct lswitch *sw, struct rconn *rconn, void *opp_)
 493 {
 494     const struct ofp_phy_port *opp = opp_;
 495     uint16_t port_no = ntohs(opp->port_no);
 496     if (sw->capabilities & OFPC_STP && port_no < STP_MAX_PORTS) {
 497         uint32_t config = ntohl(opp->config);
 498         uint32_t state = ntohl(opp->state);
 499         unsigned int *port_state = &sw->port_states[port_no];
 500         unsigned int new_port_state;
 501
 502         if (!(config & (OFPPC_NO_STP | OFPPC_PORT_DOWN))
 503             && !(state & OFPPS_LINK_DOWN))
 504         {
 505             switch (state & OFPPS_STP_MASK) {
 506             case OFPPS_STP_LISTEN:
 507                 new_port_state = P_LISTENING;
 508                 break;
 509             case OFPPS_STP_LEARN:
 510                 new_port_state = P_LEARNING;
 511                 break;
 512             case OFPPS_STP_FORWARD:
 513                 new_port_state = P_FORWARDING;
 514                 break;
 515             case OFPPS_STP_BLOCK:
 516                 new_port_state = P_BLOCKING;
 517                 break;
 518             default:
 519                 new_port_state = P_DISABLED;
 520                 break;
 521             }
 522         } else {
 523             new_port_state = P_FORWARDING;
 524         }
 525         if (*port_state != new_port_state) {
 526             *port_state = new_port_state;
 527             schedule_query(sw, 1000);
 528         }
 529     }
 530 }
 531
 532 static unsigned int
 533 get_port_state(const struct lswitch *sw, uint16_t port_no)
 534 {
 535     return (port_no >= STP_MAX_PORTS || !(sw->capabilities & OFPC_STP)
 536             ? P_FORWARDING
 537             : sw->port_states[port_no]);
 538 }
 539
 540 static bool
 541 may_learn(const struct lswitch *sw, uint16_t port_no)
 542 {
 543     return get_port_state(sw, port_no) & (P_LEARNING | P_FORWARDING);
 544 }
 545
 546 static bool
 547 may_recv(const struct lswitch *sw, uint16_t port_no, bool any_actions)
 548 {
 549     unsigned int state = get_port_state(sw, port_no);
 550     return !(any_actions
 551              ? state & (P_DISABLED | P_LISTENING | P_BLOCKING)
 552              : state & (P_DISABLED | P_LISTENING | P_BLOCKING | P_LEARNING));
 553 }
 554
 555 static bool
 556 may_send(const struct lswitch *sw, uint16_t port_no)
 557 {
 558     return get_port_state(sw, port_no) & P_FORWARDING;
 559 }
 560
 561 static void
 562 process_flow_stats(struct lswitch *sw, struct rconn *rconn,
 563                    const struct ofp_flow_stats *ofs)
 564 {
 565     const char *end = (char *) ofs + ntohs(ofs->length);
 566     bool delete = false;
 567
 568     /* Decide to delete the flow if it matches on an STP-disabled physical
 569      * port.  But don't delete it if the flow just drops all received packets,
 570      * because that's a perfectly reasonable thing to do for disabled physical
 571      * ports. */
 572     if (!(ofs->match.wildcards & htonl(OFPFW_IN_PORT))) {
 573         if (!may_recv(sw, ntohs(ofs->match.in_port),
 574                       end > (char *) ofs->actions)) {
 575             delete = true;
 576             sw->n_no_recv++;
 577         }
 578     }
 579
 580     /* Decide to delete the flow if it forwards to an STP-disabled physical
 581      * port. */
 582     if (!delete) {
 583         const struct ofp_action_header *a;
 584         size_t len;
 585
 586         for (a = ofs->actions; (char *) a < end; a += len / 8) {
 587             uint16_t type;
 588
 589             len = ntohs(a->len);
 590             if (len > end - (char *) a) {
 591                 VLOG_DBG_RL(&rl, "%012llx: action exceeds available space "
 592                             "(%zu > %td)",
 593                             sw->datapath_id, len, end - (char *) a);
 594                 break;
 595             } else if (len % 8) {
 596                 VLOG_DBG_RL(&rl, "%012llx: action length (%zu) not multiple "
 597                             "of 8 bytes", sw->datapath_id, len);
 598                 break;
 599             }
 600
 601             type = ntohs(a->type);
 602             if (a->type == htons(OFPAT_OUTPUT)) {
 603                 struct ofp_action_output *oao = (struct ofp_action_output *) a;
 604                 if (!may_send(sw, ntohs(oao->port))) {
 605                     delete = true;
 606                     sw->n_no_send++;
 607                     break;
 608                 }
 609             }
 610         }
 611     }
 612
 613     /* Delete the flow. */
 614     if (delete) {
 615         struct ofp_flow_mod *ofm;
 616         struct ofpbuf *b;
 617
 618         ofm = make_openflow(offsetof(struct ofp_flow_mod, actions),
 619                             OFPT_FLOW_MOD, &b);
 620         ofm->match = ofs->match;
 621         ofm->command = OFPFC_DELETE_STRICT;
 622         rconn_send(rconn, b, NULL);
 623     }
 624 }
 625
 626 static void
 627 process_stats_reply(struct lswitch *sw, struct rconn *rconn, void *osr_)
 628 {
 629     struct ofp_stats_reply *osr = osr_;
 630     const uint8_t *body = osr->body;
 631     const uint8_t *pos = body;
 632     size_t body_len;
 633
 634     if (sw->last_query == LLONG_MIN
 635         || osr->type != htons(OFPST_FLOW)
 636         || osr->header.xid != sw->query_xid) {
 637         return;
 638     }
 639     body_len = (ntohs(osr->header.length)
 640                 - offsetof(struct ofp_stats_reply, body));
 641     for (;;) {
 642         const struct ofp_flow_stats *fs;
 643         ptrdiff_t bytes_left = body + body_len - pos;
 644         size_t length;
 645
 646         if (bytes_left < sizeof *fs) {
 647             if (bytes_left != 0) {
 648                 VLOG_WARN_RL(&rl, "%012llx: %td leftover bytes in flow "
 649                              "stats reply", sw->datapath_id, bytes_left);
 650             }
 651             break;
 652         }
 653
 654         fs = (const void *) pos;
 655         length = ntohs(fs->length);
 656         if (length < sizeof *fs) {
 657             VLOG_WARN_RL(&rl, "%012llx: flow stats length %zu is shorter than "
 658                          "min %zu", sw->datapath_id, length, sizeof *fs);
 659             break;
 660         } else if (length > bytes_left) {
 661             VLOG_WARN_RL(&rl, "%012llx: flow stats length %zu but only %td "
 662                          "bytes left", sw->datapath_id, length, bytes_left);
 663             break;
 664         } else if ((length - sizeof *fs) % sizeof fs->actions[0]) {
 665             VLOG_WARN_RL(&rl, "%012llx: flow stats length %zu has %zu bytes "
 666                          "left over in final action", sw->datapath_id, length,
 667                          (length - sizeof *fs) % sizeof fs->actions[0]);
 668             break;
 669         }
 670
 671         sw->n_flows++;
 672         process_flow_stats(sw, rconn, fs);
 673
 674         pos += length;
 675      }
 676     if (!(osr->flags & htons(OFPSF_REPLY_MORE))) {
 677         VLOG_DBG("%012llx: Deleted %d of %d received flows to "
 678                  "implement STP, %d because of no-recv, %d because of "
 679                  "no-send", sw->datapath_id,
 680                  sw->n_no_recv + sw->n_no_send, sw->n_flows,
 681                  sw->n_no_recv, sw->n_no_send);
 682         sw->last_query = LLONG_MIN;
 683         sw->last_reply = LLONG_MIN;
 684     } else {
 685         sw->last_reply = time_msec();
 686     }
 687 }
 688