1 /* 2 * sch_plug.c Queue traffic until an explicit release command 3 * 4 * This program is free software; you can redistribute it and/or 5 * modify it under the terms of the GNU General Public License 6 * as published by the Free Software Foundation; either version 7 * 2 of the License, or (at your option) any later version. 8 * 9 * There are two ways to use this qdisc: 10 * 1. A simple "instantaneous" plug/unplug operation, by issuing an alternating 11 * sequence of TCQ_PLUG_BUFFER & TCQ_PLUG_RELEASE_INDEFINITE commands. 12 * 13 * 2. For network output buffering (a.k.a output commit) functionality. 14 * Output commit property is commonly used by applications using checkpoint 15 * based fault-tolerance to ensure that the checkpoint from which a system 16 * is being restored is consistent w.r.t outside world. 17 * 18 * Consider for e.g. Remus - a Virtual Machine checkpointing system, 19 * wherein a VM is checkpointed, say every 50ms. The checkpoint is replicated 20 * asynchronously to the backup host, while the VM continues executing the 21 * next epoch speculatively. 22 * 23 * The following is a typical sequence of output buffer operations: 24 * 1.At epoch i, start_buffer(i) 25 * 2. At end of epoch i (i.e. after 50ms): 26 * 2.1 Stop VM and take checkpoint(i). 27 * 2.2 start_buffer(i+1) and Resume VM 28 * 3. While speculatively executing epoch(i+1), asynchronously replicate 29 * checkpoint(i) to backup host. 30 * 4. When checkpoint_ack(i) is received from backup, release_buffer(i) 31 * Thus, this Qdisc would receive the following sequence of commands: 32 * TCQ_PLUG_BUFFER (epoch i) 33 * .. TCQ_PLUG_BUFFER (epoch i+1) 34 * ....TCQ_PLUG_RELEASE_ONE (epoch i) 35 * ......TCQ_PLUG_BUFFER (epoch i+2) 36 * ........ 37 */ 38 39 #include <linux/module.h> 40 #include <linux/types.h> 41 #include <linux/kernel.h> 42 #include <linux/errno.h> 43 #include <linux/netdevice.h> 44 #include <linux/skbuff.h> 45 #include <net/pkt_sched.h> 46 47 /* 48 * State of the queue, when used for network output buffering: 49 * 50 * plug(i+1) plug(i) head 51 * ------------------+--------------------+----------------> 52 * | | 53 * | | 54 * pkts_current_epoch| pkts_last_epoch |pkts_to_release 55 * ----------------->|<--------+--------->|+---------------> 56 * v v 57 * 58 */ 59 60 struct plug_sched_data { 61 /* If true, the dequeue function releases all packets 62 * from head to end of the queue. The queue turns into 63 * a pass-through queue for newly arriving packets. 64 */ 65 bool unplug_indefinite; 66 67 bool throttled; 68 69 /* Queue Limit in bytes */ 70 u32 limit; 71 72 /* Number of packets (output) from the current speculatively 73 * executing epoch. 74 */ 75 u32 pkts_current_epoch; 76 77 /* Number of packets corresponding to the recently finished 78 * epoch. These will be released when we receive a 79 * TCQ_PLUG_RELEASE_ONE command. This command is typically 80 * issued after committing a checkpoint at the target. 81 */ 82 u32 pkts_last_epoch; 83 84 /* 85 * Number of packets from the head of the queue, that can 86 * be released (committed checkpoint). 87 */ 88 u32 pkts_to_release; 89 }; 90 91 static int plug_enqueue(struct sk_buff *skb, struct Qdisc *sch, 92 struct sk_buff **to_free) 93 { 94 struct plug_sched_data *q = qdisc_priv(sch); 95 96 if (likely(sch->qstats.backlog + skb->len <= q->limit)) { 97 if (!q->unplug_indefinite) 98 q->pkts_current_epoch++; 99 return qdisc_enqueue_tail(skb, sch); 100 } 101 102 return qdisc_drop(skb, sch, to_free); 103 } 104 105 static struct sk_buff *plug_dequeue(struct Qdisc *sch) 106 { 107 struct plug_sched_data *q = qdisc_priv(sch); 108 109 if (q->throttled) 110 return NULL; 111 112 if (!q->unplug_indefinite) { 113 if (!q->pkts_to_release) { 114 /* No more packets to dequeue. Block the queue 115 * and wait for the next release command. 116 */ 117 q->throttled = true; 118 return NULL; 119 } 120 q->pkts_to_release--; 121 } 122 123 return qdisc_dequeue_head(sch); 124 } 125 126 static int plug_init(struct Qdisc *sch, struct nlattr *opt) 127 { 128 struct plug_sched_data *q = qdisc_priv(sch); 129 130 q->pkts_current_epoch = 0; 131 q->pkts_last_epoch = 0; 132 q->pkts_to_release = 0; 133 q->unplug_indefinite = false; 134 135 if (opt == NULL) { 136 q->limit = qdisc_dev(sch)->tx_queue_len 137 * psched_mtu(qdisc_dev(sch)); 138 } else { 139 struct tc_plug_qopt *ctl = nla_data(opt); 140 141 if (nla_len(opt) < sizeof(*ctl)) 142 return -EINVAL; 143 144 q->limit = ctl->limit; 145 } 146 147 q->throttled = true; 148 return 0; 149 } 150 151 /* Receives 4 types of messages: 152 * TCQ_PLUG_BUFFER: Inset a plug into the queue and 153 * buffer any incoming packets 154 * TCQ_PLUG_RELEASE_ONE: Dequeue packets from queue head 155 * to beginning of the next plug. 156 * TCQ_PLUG_RELEASE_INDEFINITE: Dequeue all packets from queue. 157 * Stop buffering packets until the next TCQ_PLUG_BUFFER 158 * command is received (just act as a pass-thru queue). 159 * TCQ_PLUG_LIMIT: Increase/decrease queue size 160 */ 161 static int plug_change(struct Qdisc *sch, struct nlattr *opt) 162 { 163 struct plug_sched_data *q = qdisc_priv(sch); 164 struct tc_plug_qopt *msg; 165 166 if (opt == NULL) 167 return -EINVAL; 168 169 msg = nla_data(opt); 170 if (nla_len(opt) < sizeof(*msg)) 171 return -EINVAL; 172 173 switch (msg->action) { 174 case TCQ_PLUG_BUFFER: 175 /* Save size of the current buffer */ 176 q->pkts_last_epoch = q->pkts_current_epoch; 177 q->pkts_current_epoch = 0; 178 if (q->unplug_indefinite) 179 q->throttled = true; 180 q->unplug_indefinite = false; 181 break; 182 case TCQ_PLUG_RELEASE_ONE: 183 /* Add packets from the last complete buffer to the 184 * packets to be released set. 185 */ 186 q->pkts_to_release += q->pkts_last_epoch; 187 q->pkts_last_epoch = 0; 188 q->throttled = false; 189 netif_schedule_queue(sch->dev_queue); 190 break; 191 case TCQ_PLUG_RELEASE_INDEFINITE: 192 q->unplug_indefinite = true; 193 q->pkts_to_release = 0; 194 q->pkts_last_epoch = 0; 195 q->pkts_current_epoch = 0; 196 q->throttled = false; 197 netif_schedule_queue(sch->dev_queue); 198 break; 199 case TCQ_PLUG_LIMIT: 200 /* Limit is supplied in bytes */ 201 q->limit = msg->limit; 202 break; 203 default: 204 return -EINVAL; 205 } 206 207 return 0; 208 } 209 210 static struct Qdisc_ops plug_qdisc_ops __read_mostly = { 211 .id = "plug", 212 .priv_size = sizeof(struct plug_sched_data), 213 .enqueue = plug_enqueue, 214 .dequeue = plug_dequeue, 215 .peek = qdisc_peek_head, 216 .init = plug_init, 217 .change = plug_change, 218 .reset = qdisc_reset_queue, 219 .owner = THIS_MODULE, 220 }; 221 222 static int __init plug_module_init(void) 223 { 224 return register_qdisc(&plug_qdisc_ops); 225 } 226 227 static void __exit plug_module_exit(void) 228 { 229 unregister_qdisc(&plug_qdisc_ops); 230 } 231 module_init(plug_module_init) 232 module_exit(plug_module_exit) 233 MODULE_LICENSE("GPL"); 234