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