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