1
2
3
4
5
6
7
8
9
10
11
12
13
14#include "qemu/osdep.h"
15#include "trace.h"
16#include "block/block_int.h"
17#include "block/blockjob_int.h"
18#include "qapi/error.h"
19#include "qapi/qmp/qerror.h"
20#include "qemu/ratelimit.h"
21#include "sysemu/block-backend.h"
22
23enum {
24
25
26
27
28
29 STREAM_BUFFER_SIZE = 512 * 1024,
30};
31
32typedef struct StreamBlockJob {
33 BlockJob common;
34 BlockDriverState *base;
35 BlockdevOnError on_error;
36 char *backing_file_str;
37 bool bs_read_only;
38 bool chain_frozen;
39} StreamBlockJob;
40
41static int coroutine_fn stream_populate(BlockBackend *blk,
42 int64_t offset, uint64_t bytes,
43 void *buf)
44{
45 QEMUIOVector qiov = QEMU_IOVEC_INIT_BUF(qiov, buf, bytes);
46
47 assert(bytes < SIZE_MAX);
48
49
50 return blk_co_preadv(blk, offset, qiov.size, &qiov, BDRV_REQ_COPY_ON_READ);
51}
52
53static void stream_abort(Job *job)
54{
55 StreamBlockJob *s = container_of(job, StreamBlockJob, common.job);
56
57 if (s->chain_frozen) {
58 BlockJob *bjob = &s->common;
59 bdrv_unfreeze_backing_chain(blk_bs(bjob->blk), s->base);
60 }
61}
62
63static int stream_prepare(Job *job)
64{
65 StreamBlockJob *s = container_of(job, StreamBlockJob, common.job);
66 BlockJob *bjob = &s->common;
67 BlockDriverState *bs = blk_bs(bjob->blk);
68 BlockDriverState *base = s->base;
69 Error *local_err = NULL;
70 int ret = 0;
71
72 bdrv_unfreeze_backing_chain(bs, base);
73 s->chain_frozen = false;
74
75 if (bs->backing) {
76 const char *base_id = NULL, *base_fmt = NULL;
77 if (base) {
78 base_id = s->backing_file_str;
79 if (base->drv) {
80 base_fmt = base->drv->format_name;
81 }
82 }
83 ret = bdrv_change_backing_file(bs, base_id, base_fmt);
84 bdrv_set_backing_hd(bs, base, &local_err);
85 if (local_err) {
86 error_report_err(local_err);
87 return -EPERM;
88 }
89 }
90
91 return ret;
92}
93
94static void stream_clean(Job *job)
95{
96 StreamBlockJob *s = container_of(job, StreamBlockJob, common.job);
97 BlockJob *bjob = &s->common;
98 BlockDriverState *bs = blk_bs(bjob->blk);
99
100
101 if (s->bs_read_only) {
102
103 blk_set_perm(bjob->blk, 0, BLK_PERM_ALL, &error_abort);
104 bdrv_reopen_set_read_only(bs, true, NULL);
105 }
106
107 g_free(s->backing_file_str);
108}
109
110static int coroutine_fn stream_run(Job *job, Error **errp)
111{
112 StreamBlockJob *s = container_of(job, StreamBlockJob, common.job);
113 BlockBackend *blk = s->common.blk;
114 BlockDriverState *bs = blk_bs(blk);
115 BlockDriverState *base = s->base;
116 int64_t len;
117 int64_t offset = 0;
118 uint64_t delay_ns = 0;
119 int error = 0;
120 int ret = 0;
121 int64_t n = 0;
122 void *buf;
123
124 if (!bs->backing) {
125 goto out;
126 }
127
128 len = bdrv_getlength(bs);
129 if (len < 0) {
130 ret = len;
131 goto out;
132 }
133 job_progress_set_remaining(&s->common.job, len);
134
135 buf = qemu_blockalign(bs, STREAM_BUFFER_SIZE);
136
137
138
139
140
141
142 if (!base) {
143 bdrv_enable_copy_on_read(bs);
144 }
145
146 for ( ; offset < len; offset += n) {
147 bool copy;
148
149
150
151
152 job_sleep_ns(&s->common.job, delay_ns);
153 if (job_is_cancelled(&s->common.job)) {
154 break;
155 }
156
157 copy = false;
158
159 ret = bdrv_is_allocated(bs, offset, STREAM_BUFFER_SIZE, &n);
160 if (ret == 1) {
161
162 } else if (ret >= 0) {
163
164
165 ret = bdrv_is_allocated_above(backing_bs(bs), base,
166 offset, n, &n);
167
168
169 if (ret == 0 && n == 0) {
170 n = len - offset;
171 }
172
173 copy = (ret == 1);
174 }
175 trace_stream_one_iteration(s, offset, n, ret);
176 if (copy) {
177 ret = stream_populate(blk, offset, n, buf);
178 }
179 if (ret < 0) {
180 BlockErrorAction action =
181 block_job_error_action(&s->common, s->on_error, true, -ret);
182 if (action == BLOCK_ERROR_ACTION_STOP) {
183 n = 0;
184 continue;
185 }
186 if (error == 0) {
187 error = ret;
188 }
189 if (action == BLOCK_ERROR_ACTION_REPORT) {
190 break;
191 }
192 }
193 ret = 0;
194
195
196 job_progress_update(&s->common.job, n);
197 if (copy) {
198 delay_ns = block_job_ratelimit_get_delay(&s->common, n);
199 } else {
200 delay_ns = 0;
201 }
202 }
203
204 if (!base) {
205 bdrv_disable_copy_on_read(bs);
206 }
207
208
209 ret = error;
210
211 qemu_vfree(buf);
212
213out:
214
215 return ret;
216}
217
218static const BlockJobDriver stream_job_driver = {
219 .job_driver = {
220 .instance_size = sizeof(StreamBlockJob),
221 .job_type = JOB_TYPE_STREAM,
222 .free = block_job_free,
223 .run = stream_run,
224 .prepare = stream_prepare,
225 .abort = stream_abort,
226 .clean = stream_clean,
227 .user_resume = block_job_user_resume,
228 .drain = block_job_drain,
229 },
230};
231
232void stream_start(const char *job_id, BlockDriverState *bs,
233 BlockDriverState *base, const char *backing_file_str,
234 int creation_flags, int64_t speed,
235 BlockdevOnError on_error, Error **errp)
236{
237 StreamBlockJob *s;
238 BlockDriverState *iter;
239 bool bs_read_only;
240
241 if (bdrv_freeze_backing_chain(bs, base, errp) < 0) {
242 return;
243 }
244
245
246 bs_read_only = bdrv_is_read_only(bs);
247 if (bs_read_only) {
248 if (bdrv_reopen_set_read_only(bs, false, errp) != 0) {
249 bs_read_only = false;
250 goto fail;
251 }
252 }
253
254
255
256
257 s = block_job_create(job_id, &stream_job_driver, NULL, bs,
258 BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED |
259 BLK_PERM_GRAPH_MOD,
260 BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED |
261 BLK_PERM_WRITE,
262 speed, creation_flags, NULL, NULL, errp);
263 if (!s) {
264 goto fail;
265 }
266
267
268
269
270
271 for (iter = backing_bs(bs); iter && iter != base; iter = backing_bs(iter)) {
272 block_job_add_bdrv(&s->common, "intermediate node", iter, 0,
273 BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED,
274 &error_abort);
275 }
276
277 s->base = base;
278 s->backing_file_str = g_strdup(backing_file_str);
279 s->bs_read_only = bs_read_only;
280 s->chain_frozen = true;
281
282 s->on_error = on_error;
283 trace_stream_start(bs, base, s);
284 job_start(&s->common.job);
285 return;
286
287fail:
288 if (bs_read_only) {
289 bdrv_reopen_set_read_only(bs, true, NULL);
290 }
291 bdrv_unfreeze_backing_chain(bs, base);
292}
293