Merge remote-tracking branch 'remotes/stefanha/tags/block-pull-request' into staging
[qemu.git] / net / queue.c
1 /*
2 * Copyright (c) 2003-2008 Fabrice Bellard
3 * Copyright (c) 2009 Red Hat, Inc.
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to deal
7 * in the Software without restriction, including without limitation the rights
8 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 * copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
18 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 * THE SOFTWARE.
22 */
23
24 #include "qemu/osdep.h"
25 #include "net/queue.h"
26 #include "qemu/queue.h"
27 #include "net/net.h"
28
29 /* The delivery handler may only return zero if it will call
30 * qemu_net_queue_flush() when it determines that it is once again able
31 * to deliver packets. It must also call qemu_net_queue_purge() in its
32 * cleanup path.
33 *
34 * If a sent callback is provided to send(), the caller must handle a
35 * zero return from the delivery handler by not sending any more packets
36 * until we have invoked the callback. Only in that case will we queue
37 * the packet.
38 *
39 * If a sent callback isn't provided, we just drop the packet to avoid
40 * unbounded queueing.
41 */
42
43 struct NetPacket {
44 QTAILQ_ENTRY(NetPacket) entry;
45 NetClientState *sender;
46 unsigned flags;
47 int size;
48 NetPacketSent *sent_cb;
49 uint8_t data[0];
50 };
51
52 struct NetQueue {
53 void *opaque;
54 uint32_t nq_maxlen;
55 uint32_t nq_count;
56 NetQueueDeliverFunc *deliver;
57
58 QTAILQ_HEAD(packets, NetPacket) packets;
59
60 unsigned delivering : 1;
61 };
62
63 NetQueue *qemu_new_net_queue(NetQueueDeliverFunc *deliver, void *opaque)
64 {
65 NetQueue *queue;
66
67 queue = g_new0(NetQueue, 1);
68
69 queue->opaque = opaque;
70 queue->nq_maxlen = 10000;
71 queue->nq_count = 0;
72 queue->deliver = deliver;
73
74 QTAILQ_INIT(&queue->packets);
75
76 queue->delivering = 0;
77
78 return queue;
79 }
80
81 void qemu_del_net_queue(NetQueue *queue)
82 {
83 NetPacket *packet, *next;
84
85 QTAILQ_FOREACH_SAFE(packet, &queue->packets, entry, next) {
86 QTAILQ_REMOVE(&queue->packets, packet, entry);
87 g_free(packet);
88 }
89
90 g_free(queue);
91 }
92
93 static void qemu_net_queue_append(NetQueue *queue,
94 NetClientState *sender,
95 unsigned flags,
96 const uint8_t *buf,
97 size_t size,
98 NetPacketSent *sent_cb)
99 {
100 NetPacket *packet;
101
102 if (queue->nq_count >= queue->nq_maxlen && !sent_cb) {
103 return; /* drop if queue full and no callback */
104 }
105 packet = g_malloc(sizeof(NetPacket) + size);
106 packet->sender = sender;
107 packet->flags = flags;
108 packet->size = size;
109 packet->sent_cb = sent_cb;
110 memcpy(packet->data, buf, size);
111
112 queue->nq_count++;
113 QTAILQ_INSERT_TAIL(&queue->packets, packet, entry);
114 }
115
116 void qemu_net_queue_append_iov(NetQueue *queue,
117 NetClientState *sender,
118 unsigned flags,
119 const struct iovec *iov,
120 int iovcnt,
121 NetPacketSent *sent_cb)
122 {
123 NetPacket *packet;
124 size_t max_len = 0;
125 int i;
126
127 if (queue->nq_count >= queue->nq_maxlen && !sent_cb) {
128 return; /* drop if queue full and no callback */
129 }
130 for (i = 0; i < iovcnt; i++) {
131 max_len += iov[i].iov_len;
132 }
133
134 packet = g_malloc(sizeof(NetPacket) + max_len);
135 packet->sender = sender;
136 packet->sent_cb = sent_cb;
137 packet->flags = flags;
138 packet->size = 0;
139
140 for (i = 0; i < iovcnt; i++) {
141 size_t len = iov[i].iov_len;
142
143 memcpy(packet->data + packet->size, iov[i].iov_base, len);
144 packet->size += len;
145 }
146
147 queue->nq_count++;
148 QTAILQ_INSERT_TAIL(&queue->packets, packet, entry);
149 }
150
151 static ssize_t qemu_net_queue_deliver(NetQueue *queue,
152 NetClientState *sender,
153 unsigned flags,
154 const uint8_t *data,
155 size_t size)
156 {
157 ssize_t ret = -1;
158 struct iovec iov = {
159 .iov_base = (void *)data,
160 .iov_len = size
161 };
162
163 queue->delivering = 1;
164 ret = queue->deliver(sender, flags, &iov, 1, queue->opaque);
165 queue->delivering = 0;
166
167 return ret;
168 }
169
170 static ssize_t qemu_net_queue_deliver_iov(NetQueue *queue,
171 NetClientState *sender,
172 unsigned flags,
173 const struct iovec *iov,
174 int iovcnt)
175 {
176 ssize_t ret = -1;
177
178 queue->delivering = 1;
179 ret = queue->deliver(sender, flags, iov, iovcnt, queue->opaque);
180 queue->delivering = 0;
181
182 return ret;
183 }
184
185 ssize_t qemu_net_queue_send(NetQueue *queue,
186 NetClientState *sender,
187 unsigned flags,
188 const uint8_t *data,
189 size_t size,
190 NetPacketSent *sent_cb)
191 {
192 ssize_t ret;
193
194 if (queue->delivering || !qemu_can_send_packet(sender)) {
195 qemu_net_queue_append(queue, sender, flags, data, size, sent_cb);
196 return 0;
197 }
198
199 ret = qemu_net_queue_deliver(queue, sender, flags, data, size);
200 if (ret == 0) {
201 qemu_net_queue_append(queue, sender, flags, data, size, sent_cb);
202 return 0;
203 }
204
205 qemu_net_queue_flush(queue);
206
207 return ret;
208 }
209
210 ssize_t qemu_net_queue_send_iov(NetQueue *queue,
211 NetClientState *sender,
212 unsigned flags,
213 const struct iovec *iov,
214 int iovcnt,
215 NetPacketSent *sent_cb)
216 {
217 ssize_t ret;
218
219 if (queue->delivering || !qemu_can_send_packet(sender)) {
220 qemu_net_queue_append_iov(queue, sender, flags, iov, iovcnt, sent_cb);
221 return 0;
222 }
223
224 ret = qemu_net_queue_deliver_iov(queue, sender, flags, iov, iovcnt);
225 if (ret == 0) {
226 qemu_net_queue_append_iov(queue, sender, flags, iov, iovcnt, sent_cb);
227 return 0;
228 }
229
230 qemu_net_queue_flush(queue);
231
232 return ret;
233 }
234
235 void qemu_net_queue_purge(NetQueue *queue, NetClientState *from)
236 {
237 NetPacket *packet, *next;
238
239 QTAILQ_FOREACH_SAFE(packet, &queue->packets, entry, next) {
240 if (packet->sender == from) {
241 QTAILQ_REMOVE(&queue->packets, packet, entry);
242 queue->nq_count--;
243 if (packet->sent_cb) {
244 packet->sent_cb(packet->sender, 0);
245 }
246 g_free(packet);
247 }
248 }
249 }
250
251 bool qemu_net_queue_flush(NetQueue *queue)
252 {
253 while (!QTAILQ_EMPTY(&queue->packets)) {
254 NetPacket *packet;
255 int ret;
256
257 packet = QTAILQ_FIRST(&queue->packets);
258 QTAILQ_REMOVE(&queue->packets, packet, entry);
259 queue->nq_count--;
260
261 ret = qemu_net_queue_deliver(queue,
262 packet->sender,
263 packet->flags,
264 packet->data,
265 packet->size);
266 if (ret == 0) {
267 queue->nq_count++;
268 QTAILQ_INSERT_HEAD(&queue->packets, packet, entry);
269 return false;
270 }
271
272 if (packet->sent_cb) {
273 packet->sent_cb(packet->sender, ret);
274 }
275
276 g_free(packet);
277 }
278 return true;
279 }