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 { 93 struct plug_sched_data *q = qdisc_priv(sch); 94 95 if (likely(sch->qstats.backlog + skb->len <= q->limit)) { 96 if (!q->unplug_indefinite) 97 q->pkts_current_epoch++; 98 return qdisc_enqueue_tail(skb, sch); 99 } 100 101 return qdisc_drop(skb, sch); 102 } 103 104 static struct sk_buff *plug_dequeue(struct Qdisc *sch) 105 { 106 struct plug_sched_data *q = qdisc_priv(sch); 107 108 if (q->throttled) 109 return NULL; 110 111 if (!q->unplug_indefinite) { 112 if (!q->pkts_to_release) { 113 /* No more packets to dequeue. Block the queue 114 * and wait for the next release command. 115 */ 116 q->throttled = true; 117 return NULL; 118 } 119 q->pkts_to_release--; 120 } 121 122 return qdisc_dequeue_head(sch); 123 } 124 125 static int plug_init(struct Qdisc *sch, struct nlattr *opt) 126 { 127 struct plug_sched_data *q = qdisc_priv(sch); 128 129 q->pkts_current_epoch = 0; 130 q->pkts_last_epoch = 0; 131 q->pkts_to_release = 0; 132 q->unplug_indefinite = false; 133 134 if (opt == NULL) { 135 q->limit = qdisc_dev(sch)->tx_queue_len 136 * psched_mtu(qdisc_dev(sch)); 137 } else { 138 struct tc_plug_qopt *ctl = nla_data(opt); 139 140 if (nla_len(opt) < sizeof(*ctl)) 141 return -EINVAL; 142 143 q->limit = ctl->limit; 144 } 145 146 q->throttled = true; 147 return 0; 148 } 149 150 /* Receives 4 types of messages: 151 * TCQ_PLUG_BUFFER: Inset a plug into the queue and 152 * buffer any incoming packets 153 * TCQ_PLUG_RELEASE_ONE: Dequeue packets from queue head 154 * to beginning of the next plug. 155 * TCQ_PLUG_RELEASE_INDEFINITE: Dequeue all packets from queue. 156 * Stop buffering packets until the next TCQ_PLUG_BUFFER 157 * command is received (just act as a pass-thru queue). 158 * TCQ_PLUG_LIMIT: Increase/decrease queue size 159 */ 160 static int plug_change(struct Qdisc *sch, struct nlattr *opt) 161 { 162 struct plug_sched_data *q = qdisc_priv(sch); 163 struct tc_plug_qopt *msg; 164 165 if (opt == NULL) 166 return -EINVAL; 167 168 msg = nla_data(opt); 169 if (nla_len(opt) < sizeof(*msg)) 170 return -EINVAL; 171 172 switch (msg->action) { 173 case TCQ_PLUG_BUFFER: 174 /* Save size of the current buffer */ 175 q->pkts_last_epoch = q->pkts_current_epoch; 176 q->pkts_current_epoch = 0; 177 if (q->unplug_indefinite) 178 q->throttled = true; 179 q->unplug_indefinite = false; 180 break; 181 case TCQ_PLUG_RELEASE_ONE: 182 /* Add packets from the last complete buffer to the 183 * packets to be released set. 184 */ 185 q->pkts_to_release += q->pkts_last_epoch; 186 q->pkts_last_epoch = 0; 187 q->throttled = false; 188 netif_schedule_queue(sch->dev_queue); 189 break; 190 case TCQ_PLUG_RELEASE_INDEFINITE: 191 q->unplug_indefinite = true; 192 q->pkts_to_release = 0; 193 q->pkts_last_epoch = 0; 194 q->pkts_current_epoch = 0; 195 q->throttled = false; 196 netif_schedule_queue(sch->dev_queue); 197 break; 198 case TCQ_PLUG_LIMIT: 199 /* Limit is supplied in bytes */ 200 q->limit = msg->limit; 201 break; 202 default: 203 return -EINVAL; 204 } 205 206 return 0; 207 } 208 209 static struct Qdisc_ops plug_qdisc_ops __read_mostly = { 210 .id = "plug", 211 .priv_size = sizeof(struct plug_sched_data), 212 .enqueue = plug_enqueue, 213 .dequeue = plug_dequeue, 214 .peek = qdisc_peek_head, 215 .init = plug_init, 216 .change = plug_change, 217 .reset = qdisc_reset_queue, 218 .owner = THIS_MODULE, 219 }; 220 221 static int __init plug_module_init(void) 222 { 223 return register_qdisc(&plug_qdisc_ops); 224 } 225 226 static void __exit plug_module_exit(void) 227 { 228 unregister_qdisc(&plug_qdisc_ops); 229 } 230 module_init(plug_module_init) 231 module_exit(plug_module_exit) 232 MODULE_LICENSE("GPL"); 233