]> git.proxmox.com Git - mirror_qemu.git/blame - buffered_file.c
block: add close notifiers
[mirror_qemu.git] / buffered_file.c
CommitLineData
39b65c2e
AL
1/*
2 * QEMU buffered QEMUFile
3 *
4 * Copyright IBM, Corp. 2008
5 *
6 * Authors:
7 * Anthony Liguori <aliguori@us.ibm.com>
8 *
9 * This work is licensed under the terms of the GNU GPL, version 2. See
10 * the COPYING file in the top-level directory.
11 *
6b620ca3
PB
12 * Contributions after 2012-01-13 are licensed under the terms of the
13 * GNU GPL, version 2 or (at your option) any later version.
39b65c2e
AL
14 */
15
16#include "qemu-common.h"
17#include "hw/hw.h"
18#include "qemu-timer.h"
39b65c2e
AL
19#include "qemu-char.h"
20#include "buffered_file.h"
21
22//#define DEBUG_BUFFERED_FILE
23
24typedef struct QEMUFileBuffered
25{
c7a8f0cd 26 MigrationState *migration_state;
39b65c2e 27 QEMUFile *file;
39b65c2e
AL
28 int freeze_output;
29 size_t bytes_xfer;
30 size_t xfer_limit;
31 uint8_t *buffer;
32 size_t buffer_size;
33 size_t buffer_capacity;
34 QEMUTimer *timer;
35} QEMUFileBuffered;
36
37#ifdef DEBUG_BUFFERED_FILE
d0f2c4c6 38#define DPRINTF(fmt, ...) \
39b65c2e
AL
39 do { printf("buffered-file: " fmt, ## __VA_ARGS__); } while (0)
40#else
d0f2c4c6 41#define DPRINTF(fmt, ...) \
39b65c2e
AL
42 do { } while (0)
43#endif
44
45static void buffered_append(QEMUFileBuffered *s,
46 const uint8_t *buf, size_t size)
47{
48 if (size > (s->buffer_capacity - s->buffer_size)) {
d0f2c4c6 49 DPRINTF("increasing buffer capacity from %zu by %zu\n",
39b65c2e
AL
50 s->buffer_capacity, size + 1024);
51
52 s->buffer_capacity += size + 1024;
53
05e72dc5 54 s->buffer = g_realloc(s->buffer, s->buffer_capacity);
39b65c2e
AL
55 }
56
57 memcpy(s->buffer + s->buffer_size, buf, size);
58 s->buffer_size += size;
59}
60
d2dbc8e6 61static ssize_t buffered_flush(QEMUFileBuffered *s)
39b65c2e
AL
62{
63 size_t offset = 0;
d2dbc8e6 64 ssize_t ret = 0;
39b65c2e 65
d0f2c4c6 66 DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
39b65c2e 67
2dddf6f4 68 while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
39b65c2e 69
c87b015b
JQ
70 ret = migrate_fd_put_buffer(s->migration_state, s->buffer + offset,
71 s->buffer_size - offset);
39b65c2e 72 if (ret == -EAGAIN) {
d0f2c4c6 73 DPRINTF("backend not ready, freezing\n");
d2dbc8e6 74 ret = 0;
39b65c2e
AL
75 s->freeze_output = 1;
76 break;
77 }
78
79 if (ret <= 0) {
d0f2c4c6 80 DPRINTF("error flushing data, %zd\n", ret);
39b65c2e
AL
81 break;
82 } else {
d0f2c4c6 83 DPRINTF("flushed %zd byte(s)\n", ret);
39b65c2e 84 offset += ret;
2dddf6f4 85 s->bytes_xfer += ret;
39b65c2e
AL
86 }
87 }
88
d0f2c4c6 89 DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size);
39b65c2e
AL
90 memmove(s->buffer, s->buffer + offset, s->buffer_size - offset);
91 s->buffer_size -= offset;
d2dbc8e6
JQ
92
93 if (ret < 0) {
94 return ret;
95 }
96 return offset;
39b65c2e
AL
97}
98
99static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
100{
101 QEMUFileBuffered *s = opaque;
d2dbc8e6 102 ssize_t error;
39b65c2e 103
d0f2c4c6 104 DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
39b65c2e 105
42802d47
JQ
106 error = qemu_file_get_error(s->file);
107 if (error) {
108 DPRINTF("flush when error, bailing: %s\n", strerror(-error));
109 return error;
39b65c2e
AL
110 }
111
d0f2c4c6 112 DPRINTF("unfreezing output\n");
39b65c2e
AL
113 s->freeze_output = 0;
114
2dddf6f4 115 if (size > 0) {
d0f2c4c6 116 DPRINTF("buffering %d bytes\n", size - offset);
2dddf6f4 117 buffered_append(s, buf, size);
39b65c2e
AL
118 }
119
d2dbc8e6
JQ
120 error = buffered_flush(s);
121 if (error < 0) {
122 DPRINTF("buffered flush error. bailing: %s\n", strerror(-error));
d2dbc8e6
JQ
123 return error;
124 }
2dddf6f4 125
5e77aaa0
AK
126 if (pos == 0 && size == 0) {
127 DPRINTF("file is ready\n");
2dddf6f4 128 if (!s->freeze_output && s->bytes_xfer < s->xfer_limit) {
5e77aaa0 129 DPRINTF("notifying client\n");
2c9adcb8 130 migrate_fd_put_ready(s->migration_state);
5e77aaa0
AK
131 }
132 }
133
2dddf6f4 134 return size;
39b65c2e
AL
135}
136
137static int buffered_close(void *opaque)
138{
139 QEMUFileBuffered *s = opaque;
d2dbc8e6
JQ
140 ssize_t ret = 0;
141 int ret2;
39b65c2e 142
d0f2c4c6 143 DPRINTF("closing\n");
39b65c2e 144
2dddf6f4 145 s->xfer_limit = INT_MAX;
624b9cc2 146 while (!qemu_file_get_error(s->file) && s->buffer_size) {
d2dbc8e6
JQ
147 ret = buffered_flush(s);
148 if (ret < 0) {
149 break;
150 }
9499743f
JQ
151 if (s->freeze_output) {
152 ret = migrate_fd_wait_for_unfreeze(s->migration_state);
153 if (ret < 0) {
154 break;
155 }
156 }
39b65c2e
AL
157 }
158
d2dbc8e6
JQ
159 ret2 = migrate_fd_close(s->migration_state);
160 if (ret >= 0) {
161 ret = ret2;
162 }
39b65c2e
AL
163 qemu_del_timer(s->timer);
164 qemu_free_timer(s->timer);
7267c094
AL
165 g_free(s->buffer);
166 g_free(s);
39b65c2e
AL
167
168 return ret;
169}
170
4fc7d819
JQ
171/*
172 * The meaning of the return values is:
173 * 0: We can continue sending
174 * 1: Time to stop
42802d47 175 * negative: There has been an error
4fc7d819 176 */
39b65c2e
AL
177static int buffered_rate_limit(void *opaque)
178{
179 QEMUFileBuffered *s = opaque;
42802d47 180 int ret;
39b65c2e 181
42802d47
JQ
182 ret = qemu_file_get_error(s->file);
183 if (ret) {
184 return ret;
4fc7d819 185 }
39b65c2e
AL
186 if (s->freeze_output)
187 return 1;
188
189 if (s->bytes_xfer > s->xfer_limit)
190 return 1;
191
192 return 0;
193}
194
3d002df3 195static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
19629537
GC
196{
197 QEMUFileBuffered *s = opaque;
624b9cc2 198 if (qemu_file_get_error(s->file)) {
19629537 199 goto out;
624b9cc2 200 }
3d002df3
MT
201 if (new_rate > SIZE_MAX) {
202 new_rate = SIZE_MAX;
203 }
204
19629537
GC
205 s->xfer_limit = new_rate / 10;
206
207out:
208 return s->xfer_limit;
209}
210
3d002df3 211static int64_t buffered_get_rate_limit(void *opaque)
c163b5ca
LS
212{
213 QEMUFileBuffered *s = opaque;
214
215 return s->xfer_limit;
216}
217
39b65c2e
AL
218static void buffered_rate_tick(void *opaque)
219{
220 QEMUFileBuffered *s = opaque;
221
624b9cc2 222 if (qemu_file_get_error(s->file)) {
e447b1a6 223 buffered_close(s);
39b65c2e 224 return;
e447b1a6 225 }
39b65c2e 226
7bd427d8 227 qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
39b65c2e
AL
228
229 if (s->freeze_output)
230 return;
231
232 s->bytes_xfer = 0;
233
2dddf6f4 234 buffered_put_buffer(s, NULL, 0, 0);
39b65c2e
AL
235}
236
796b4b0f 237QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
39b65c2e
AL
238{
239 QEMUFileBuffered *s;
240
7267c094 241 s = g_malloc0(sizeof(*s));
39b65c2e 242
c7a8f0cd 243 s->migration_state = migration_state;
796b4b0f 244 s->xfer_limit = migration_state->bandwidth_limit / 10;
39b65c2e
AL
245
246 s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
19629537 247 buffered_close, buffered_rate_limit,
c163b5ca
LS
248 buffered_set_rate_limit,
249 buffered_get_rate_limit);
39b65c2e 250
7bd427d8 251 s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
39b65c2e 252
7bd427d8 253 qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
39b65c2e
AL
254
255 return s->file;
256}