ofproto/pinsched.c

   1 /*
   2  * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013 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 "pinsched.h"
  19 #include <sys/types.h>
  20 #include <netinet/in.h>
  21 #include <arpa/inet.h>
  22 #include <stdint.h>
  23 #include <stdlib.h>
  24 #include "flow.h"
  25 #include "hash.h"
  26 #include "hmap.h"
  27 #include "ofpbuf.h"
  28 #include "openflow/openflow.h"
  29 #include "poll-loop.h"
  30 #include "random.h"
  31 #include "rconn.h"
  32 #include "sat-math.h"
  33 #include "timeval.h"
  34 #include "token-bucket.h"
  35 #include "vconn.h"
  36
  37 struct pinqueue {
  38     struct hmap_node node;      /* In struct pinsched's 'queues' hmap. */
  39     ofp_port_t port_no;           /* Port number. */
  40     struct list packets;        /* Contains "struct ofpbuf"s. */
  41     int n;                      /* Number of packets in 'packets'. */
  42 };
  43
  44 struct pinsched {
  45     struct token_bucket token_bucket;
  46
  47     /* One queue per physical port. */
  48     struct hmap queues;         /* Contains "struct pinqueue"s. */
  49     int n_queued;               /* Sum over queues[*].n. */
  50     struct pinqueue *next_txq;  /* Next pinqueue check in round-robin. */
  51
  52     /* Transmission queue. */
  53     int n_txq;                  /* No. of packets waiting in rconn for tx. */
  54
  55     /* Statistics reporting. */
  56     unsigned long long n_normal;        /* # txed w/o rate limit queuing. */
  57     unsigned long long n_limited;       /* # queued for rate limiting. */
  58     unsigned long long n_queue_dropped; /* # dropped due to queue overflow. */
  59 };
  60
  61 static void
  62 advance_txq(struct pinsched *ps)
  63 {
  64     struct hmap_node *next;
  65
  66     next = (ps->next_txq
  67             ? hmap_next(&ps->queues, &ps->next_txq->node)
  68             : hmap_first(&ps->queues));
  69     ps->next_txq = next ? CONTAINER_OF(next, struct pinqueue, node) : NULL;
  70 }
  71
  72 static struct ofpbuf *
  73 dequeue_packet(struct pinsched *ps, struct pinqueue *q)
  74 {
  75     struct ofpbuf *packet = ofpbuf_from_list(list_pop_front(&q->packets));
  76     q->n--;
  77     ps->n_queued--;
  78     return packet;
  79 }
  80
  81 static void
  82 adjust_limits(int *rate_limit, int *burst_limit)
  83 {
  84     if (*rate_limit <= 0) {
  85         *rate_limit = 1000;
  86     }
  87     if (*burst_limit <= 0) {
  88         *burst_limit = *rate_limit / 4;
  89     }
  90     if (*burst_limit < 1) {
  91         *burst_limit = 1;
  92     }
  93 }
  94
  95 /* Destroys 'q' and removes it from 'ps''s set of queues.
  96  * (The caller must ensure that 'q' is empty.) */
  97 static void
  98 pinqueue_destroy(struct pinsched *ps, struct pinqueue *q)
  99 {
 100     hmap_remove(&ps->queues, &q->node);
 101     free(q);
 102 }
 103
 104 static struct pinqueue *
 105 pinqueue_get(struct pinsched *ps, ofp_port_t port_no)
 106 {
 107     uint32_t hash = hash_ofp_port(port_no);
 108     struct pinqueue *q;
 109
 110     HMAP_FOR_EACH_IN_BUCKET (q, node, hash, &ps->queues) {
 111         if (port_no == q->port_no) {
 112             return q;
 113         }
 114     }
 115
 116     q = xmalloc(sizeof *q);
 117     hmap_insert(&ps->queues, &q->node, hash);
 118     q->port_no = port_no;
 119     list_init(&q->packets);
 120     q->n = 0;
 121     return q;
 122 }
 123
 124 /* Drop a packet from the longest queue in 'ps'. */
 125 static void
 126 drop_packet(struct pinsched *ps)
 127 {
 128     struct pinqueue *longest;   /* Queue currently selected as longest. */
 129     int n_longest = 0;          /* # of queues of same length as 'longest'. */
 130     struct pinqueue *q;
 131
 132     ps->n_queue_dropped++;
 133
 134     longest = NULL;
 135     HMAP_FOR_EACH (q, node, &ps->queues) {
 136         if (!longest || longest->n < q->n) {
 137             longest = q;
 138             n_longest = 1;
 139         } else if (longest->n == q->n) {
 140             n_longest++;
 141
 142             /* Randomly select one of the longest queues, with a uniform
 143              * distribution (Knuth algorithm 3.4.2R). */
 144             if (!random_range(n_longest)) {
 145                 longest = q;
 146             }
 147         }
 148     }
 149
 150     /* FIXME: do we want to pop the tail instead? */
 151     ofpbuf_delete(dequeue_packet(ps, longest));
 152     if (longest->n == 0) {
 153         pinqueue_destroy(ps, longest);
 154     }
 155 }
 156
 157 /* Remove and return the next packet to transmit (in round-robin order). */
 158 static struct ofpbuf *
 159 get_tx_packet(struct pinsched *ps)
 160 {
 161     struct ofpbuf *packet;
 162     struct pinqueue *q;
 163
 164     if (!ps->next_txq) {
 165         advance_txq(ps);
 166     }
 167
 168     q = ps->next_txq;
 169     packet = dequeue_packet(ps, q);
 170     advance_txq(ps);
 171     if (q->n == 0) {
 172         pinqueue_destroy(ps, q);
 173     }
 174
 175     return packet;
 176 }
 177
 178 /* Attempts to remove enough tokens from 'ps' to transmit a packet.  Returns
 179  * true if successful, false otherwise.  (In the latter case no tokens are
 180  * removed.) */
 181 static bool
 182 get_token(struct pinsched *ps)
 183 {
 184     return token_bucket_withdraw(&ps->token_bucket, 1000);
 185 }
 186
 187 void
 188 pinsched_send(struct pinsched *ps, ofp_port_t port_no,
 189               struct ofpbuf *packet, struct list *txq)
 190 {
 191     list_init(txq);
 192     if (!ps) {
 193         list_push_back(txq, &packet->list_node);
 194     } else if (!ps->n_queued && get_token(ps)) {
 195         /* In the common case where we are not constrained by the rate limit,
 196          * let the packet take the normal path. */
 197         ps->n_normal++;
 198         list_push_back(txq, &packet->list_node);
 199     } else {
 200         /* Otherwise queue it up for the periodic callback to drain out. */
 201         struct pinqueue *q;
 202
 203         /* We might be called with a buffer obtained from dpif_recv() that has
 204          * much more allocated space than actual content most of the time.
 205          * Since we're going to store the packet for some time, free up that
 206          * otherwise wasted space. */
 207         ofpbuf_trim(packet);
 208
 209         if (ps->n_queued * 1000 >= ps->token_bucket.burst) {
 210             drop_packet(ps);
 211         }
 212         q = pinqueue_get(ps, port_no);
 213         list_push_back(&q->packets, &packet->list_node);
 214         q->n++;
 215         ps->n_queued++;
 216         ps->n_limited++;
 217     }
 218 }
 219
 220 void
 221 pinsched_run(struct pinsched *ps, struct list *txq)
 222 {
 223     list_init(txq);
 224     if (ps) {
 225         int i;
 226
 227         /* Drain some packets out of the bucket if possible, but limit the
 228          * number of iterations to allow other code to get work done too. */
 229         for (i = 0; ps->n_queued && get_token(ps) && i < 50; i++) {
 230             struct ofpbuf *packet = get_tx_packet(ps);
 231             list_push_back(txq, &packet->list_node);
 232         }
 233     }
 234 }
 235
 236 void
 237 pinsched_wait(struct pinsched *ps)
 238 {
 239     if (ps && ps->n_queued) {
 240         token_bucket_wait(&ps->token_bucket, 1000);
 241     }
 242 }
 243
 244 /* Creates and returns a scheduler for sending packet-in messages. */
 245 struct pinsched *
 246 pinsched_create(int rate_limit, int burst_limit)
 247 {
 248     struct pinsched *ps;
 249
 250     ps = xzalloc(sizeof *ps);
 251
 252     adjust_limits(&rate_limit, &burst_limit);
 253     token_bucket_init(&ps->token_bucket,
 254                       rate_limit, sat_mul(burst_limit, 1000));
 255
 256     hmap_init(&ps->queues);
 257     ps->n_queued = 0;
 258     ps->next_txq = NULL;
 259     ps->n_txq = 0;
 260     ps->n_normal = 0;
 261     ps->n_limited = 0;
 262     ps->n_queue_dropped = 0;
 263
 264     return ps;
 265 }
 266
 267 void
 268 pinsched_destroy(struct pinsched *ps)
 269 {
 270     if (ps) {
 271         struct pinqueue *q, *next;
 272
 273         HMAP_FOR_EACH_SAFE (q, next, node, &ps->queues) {
 274             hmap_remove(&ps->queues, &q->node);
 275             ofpbuf_list_delete(&q->packets);
 276             free(q);
 277         }
 278         hmap_destroy(&ps->queues);
 279         free(ps);
 280     }
 281 }
 282
 283 void
 284 pinsched_get_limits(const struct pinsched *ps,
 285                     int *rate_limit, int *burst_limit)
 286 {
 287     *rate_limit = ps->token_bucket.rate;
 288     *burst_limit = ps->token_bucket.burst / 1000;
 289 }
 290
 291 void
 292 pinsched_set_limits(struct pinsched *ps, int rate_limit, int burst_limit)
 293 {
 294     adjust_limits(&rate_limit, &burst_limit);
 295     token_bucket_set(&ps->token_bucket,
 296                      rate_limit, sat_mul(burst_limit, 1000));
 297     while (ps->n_queued > burst_limit) {
 298         drop_packet(ps);
 299     }
 300 }
 301
 302 /* Returns the number of packets scheduled to be sent eventually by 'ps'.
 303  * Returns 0 if 'ps' is null. */
 304 unsigned int
 305 pinsched_count_txqlen(const struct pinsched *ps)
 306 {
 307     return ps ? ps->n_txq : 0;
 308 }