Fix off-by-one in jobs_eta allocation
[fio.git] / eta.c
... / ...
CommitLineData
1/*
2 * Status and ETA code
3 */
4#include <unistd.h>
5#include <fcntl.h>
6#include <string.h>
7
8#include "fio.h"
9
10static char run_str[REAL_MAX_JOBS + 1];
11
12/*
13 * Sets the status of the 'td' in the printed status map.
14 */
15static void check_str_update(struct thread_data *td)
16{
17 char c = run_str[td->thread_number - 1];
18
19 switch (td->runstate) {
20 case TD_REAPED:
21 c = '_';
22 break;
23 case TD_EXITED:
24 c = 'E';
25 break;
26 case TD_RAMP:
27 c = '/';
28 break;
29 case TD_RUNNING:
30 if (td_rw(td)) {
31 if (td_random(td)) {
32 if (td->o.rwmix[DDIR_READ] == 100)
33 c = 'r';
34 else if (td->o.rwmix[DDIR_WRITE] == 100)
35 c = 'w';
36 else
37 c = 'm';
38 } else {
39 if (td->o.rwmix[DDIR_READ] == 100)
40 c = 'R';
41 else if (td->o.rwmix[DDIR_WRITE] == 100)
42 c = 'W';
43 else
44 c = 'M';
45 }
46 } else if (td_read(td)) {
47 if (td_random(td))
48 c = 'r';
49 else
50 c = 'R';
51 } else {
52 if (td_random(td))
53 c = 'w';
54 else
55 c = 'W';
56 }
57 break;
58 case TD_PRE_READING:
59 c = 'p';
60 break;
61 case TD_VERIFYING:
62 c = 'V';
63 break;
64 case TD_FSYNCING:
65 c = 'F';
66 break;
67 case TD_CREATED:
68 c = 'C';
69 break;
70 case TD_INITIALIZED:
71 c = 'I';
72 break;
73 case TD_NOT_CREATED:
74 c = 'P';
75 break;
76 default:
77 log_err("state %d\n", td->runstate);
78 }
79
80 run_str[td->thread_number - 1] = c;
81}
82
83/*
84 * Convert seconds to a printable string.
85 */
86static void eta_to_str(char *str, unsigned long eta_sec)
87{
88 unsigned int d, h, m, s;
89 int disp_hour = 0;
90
91 s = eta_sec % 60;
92 eta_sec /= 60;
93 m = eta_sec % 60;
94 eta_sec /= 60;
95 h = eta_sec % 24;
96 eta_sec /= 24;
97 d = eta_sec;
98
99 if (d) {
100 disp_hour = 1;
101 str += sprintf(str, "%02ud:", d);
102 }
103
104 if (h || disp_hour)
105 str += sprintf(str, "%02uh:", h);
106
107 str += sprintf(str, "%02um:", m);
108 str += sprintf(str, "%02us", s);
109}
110
111/*
112 * Best effort calculation of the estimated pending runtime of a job.
113 */
114static int thread_eta(struct thread_data *td)
115{
116 unsigned long long bytes_total, bytes_done;
117 unsigned long eta_sec = 0;
118 unsigned long elapsed;
119
120 elapsed = (mtime_since_now(&td->epoch) + 999) / 1000;
121
122 bytes_total = td->total_io_size;
123
124 if (td->o.fill_device && td->o.size == -1ULL) {
125 if (!td->fill_device_size || td->fill_device_size == -1ULL)
126 return 0;
127
128 bytes_total = td->fill_device_size;
129 }
130
131 /*
132 * if writing, bytes_total will be twice the size. If mixing,
133 * assume a 50/50 split and thus bytes_total will be 50% larger.
134 */
135 if (td->o.do_verify && td->o.verify && td_write(td)) {
136 if (td_rw(td))
137 bytes_total = bytes_total * 3 / 2;
138 else
139 bytes_total <<= 1;
140 }
141
142 if (td->o.zone_size && td->o.zone_skip)
143 bytes_total /= (td->o.zone_skip / td->o.zone_size);
144
145 if (td->runstate == TD_RUNNING || td->runstate == TD_VERIFYING) {
146 double perc, perc_t;
147
148 bytes_done = td->io_bytes[DDIR_READ] + td->io_bytes[DDIR_WRITE];
149 perc = (double) bytes_done / (double) bytes_total;
150 if (perc > 1.0)
151 perc = 1.0;
152
153 if (td->o.time_based) {
154 perc_t = (double) elapsed / (double) td->o.timeout;
155 if (perc_t < perc)
156 perc = perc_t;
157 }
158
159 eta_sec = (unsigned long) (elapsed * (1.0 / perc)) - elapsed;
160
161 if (td->o.timeout &&
162 eta_sec > (td->o.timeout + done_secs - elapsed))
163 eta_sec = td->o.timeout + done_secs - elapsed;
164 } else if (td->runstate == TD_NOT_CREATED || td->runstate == TD_CREATED
165 || td->runstate == TD_INITIALIZED
166 || td->runstate == TD_RAMP
167 || td->runstate == TD_PRE_READING) {
168 int t_eta = 0, r_eta = 0;
169
170 /*
171 * We can only guess - assume it'll run the full timeout
172 * if given, otherwise assume it'll run at the specified rate.
173 */
174 if (td->o.timeout) {
175 t_eta = td->o.timeout + td->o.start_delay +
176 td->o.ramp_time;
177
178 if (in_ramp_time(td)) {
179 unsigned long ramp_left;
180
181 ramp_left = mtime_since_now(&td->epoch);
182 ramp_left = (ramp_left + 999) / 1000;
183 if (ramp_left <= t_eta)
184 t_eta -= ramp_left;
185 }
186 }
187 if (td->o.rate[0] || td->o.rate[1]) {
188 r_eta = (bytes_total / 1024) /
189 (td->o.rate[0] + td->o.rate[1]);
190 r_eta += td->o.start_delay;
191 }
192
193 if (r_eta && t_eta)
194 eta_sec = min(r_eta, t_eta);
195 else if (r_eta)
196 eta_sec = r_eta;
197 else if (t_eta)
198 eta_sec = t_eta;
199 else
200 eta_sec = 0;
201 } else {
202 /*
203 * thread is already done or waiting for fsync
204 */
205 eta_sec = 0;
206 }
207
208 return eta_sec;
209}
210
211static void calc_rate(unsigned long mtime, unsigned long long *io_bytes,
212 unsigned long long *prev_io_bytes, unsigned int *rate)
213{
214 rate[0] = (io_bytes[0] - prev_io_bytes[0]) / mtime;
215 rate[1] = (io_bytes[1] - prev_io_bytes[1]) / mtime;
216 prev_io_bytes[0] = io_bytes[0];
217 prev_io_bytes[1] = io_bytes[1];
218}
219
220static void calc_iops(unsigned long mtime, unsigned long long *io_iops,
221 unsigned long long *prev_io_iops, unsigned int *iops)
222{
223 iops[0] = ((io_iops[0] - prev_io_iops[0]) * 1000) / mtime;
224 iops[1] = ((io_iops[1] - prev_io_iops[1]) * 1000) / mtime;
225 prev_io_iops[0] = io_iops[0];
226 prev_io_iops[1] = io_iops[1];
227}
228
229/*
230 * Print status of the jobs we know about. This includes rate estimates,
231 * ETA, thread state, etc.
232 */
233int calc_thread_status(struct jobs_eta *je, int force)
234{
235 struct thread_data *td;
236 int i;
237 unsigned long rate_time, disp_time, bw_avg_time, *eta_secs;
238 unsigned long long io_bytes[2];
239 unsigned long long io_iops[2];
240 struct timeval now;
241
242 static unsigned long long rate_io_bytes[2];
243 static unsigned long long disp_io_bytes[2];
244 static unsigned long long disp_io_iops[2];
245 static struct timeval rate_prev_time, disp_prev_time;
246 int i2p = 0;
247
248 if (!force) {
249 if (temp_stall_ts || terse_output || eta_print == FIO_ETA_NEVER)
250 return 0;
251
252 if (!isatty(STDOUT_FILENO) && (eta_print != FIO_ETA_ALWAYS))
253 return 0;
254 }
255
256 if (!rate_io_bytes[0] && !rate_io_bytes[1])
257 fill_start_time(&rate_prev_time);
258 if (!disp_io_bytes[0] && !disp_io_bytes[1])
259 fill_start_time(&disp_prev_time);
260
261 eta_secs = malloc(thread_number * sizeof(unsigned long));
262 memset(eta_secs, 0, thread_number * sizeof(unsigned long));
263
264 je->elapsed_sec = (mtime_since_genesis() + 999) / 1000;
265
266 io_bytes[0] = io_bytes[1] = 0;
267 io_iops[0] = io_iops[1] = 0;
268 bw_avg_time = ULONG_MAX;
269 for_each_td(td, i) {
270 if (td->o.bw_avg_time < bw_avg_time)
271 bw_avg_time = td->o.bw_avg_time;
272 if (td->runstate == TD_RUNNING || td->runstate == TD_VERIFYING
273 || td->runstate == TD_FSYNCING
274 || td->runstate == TD_PRE_READING) {
275 je->nr_running++;
276 je->t_rate += td->o.rate[0] + td->o.rate[1];
277 je->m_rate += td->o.ratemin[0] + td->o.ratemin[1];
278 je->t_iops += td->o.rate_iops[0] + td->o.rate_iops[1];
279 je->m_iops += td->o.rate_iops_min[0] +
280 td->o.rate_iops_min[1];
281 je->files_open += td->nr_open_files;
282 } else if (td->runstate == TD_RAMP) {
283 je->nr_running++;
284 je->nr_ramp++;
285 } else if (td->runstate < TD_RUNNING)
286 je->nr_pending++;
287
288 if (je->elapsed_sec >= 3)
289 eta_secs[i] = thread_eta(td);
290 else
291 eta_secs[i] = INT_MAX;
292
293 check_str_update(td);
294
295 if (td->runstate > TD_RAMP) {
296 io_bytes[0] += td->io_bytes[0];
297 io_bytes[1] += td->io_bytes[1];
298 io_iops[0] += td->io_blocks[0];
299 io_iops[1] += td->io_blocks[1];
300 }
301 }
302
303 if (exitall_on_terminate)
304 je->eta_sec = INT_MAX;
305 else
306 je->eta_sec = 0;
307
308 for_each_td(td, i) {
309 if (!i2p && is_power_of_2(td->o.kb_base))
310 i2p = 1;
311 if (exitall_on_terminate) {
312 if (eta_secs[i] < je->eta_sec)
313 je->eta_sec = eta_secs[i];
314 } else {
315 if (eta_secs[i] > je->eta_sec)
316 je->eta_sec = eta_secs[i];
317 }
318 }
319
320 free(eta_secs);
321
322 fio_gettime(&now, NULL);
323 rate_time = mtime_since(&rate_prev_time, &now);
324
325 if (write_bw_log && rate_time > bw_avg_time && !in_ramp_time(td)) {
326 calc_rate(rate_time, io_bytes, rate_io_bytes, je->rate);
327 memcpy(&rate_prev_time, &now, sizeof(now));
328 add_agg_sample(je->rate[DDIR_READ], DDIR_READ, 0);
329 add_agg_sample(je->rate[DDIR_WRITE], DDIR_WRITE, 0);
330 }
331
332 disp_time = mtime_since(&disp_prev_time, &now);
333
334 /*
335 * Allow a little slack, the target is to print it every 1000 msecs
336 */
337 if (!force && disp_time < 900)
338 return 0;
339
340 calc_rate(disp_time, io_bytes, disp_io_bytes, je->rate);
341 calc_iops(disp_time, io_iops, disp_io_iops, je->iops);
342
343 memcpy(&disp_prev_time, &now, sizeof(now));
344
345 if (!force && !je->nr_running && !je->nr_pending)
346 return 0;
347
348 je->nr_threads = thread_number;
349 memcpy(je->run_str, run_str, thread_number * sizeof(char));
350
351 return 1;
352}
353
354void display_thread_status(struct jobs_eta *je)
355{
356 static int linelen_last;
357 static int eta_good;
358 char output[512], *p = output;
359 char eta_str[128];
360 double perc = 0.0;
361 int i2p = 0;
362
363 if (je->eta_sec != INT_MAX && je->elapsed_sec) {
364 perc = (double) je->elapsed_sec / (double) (je->elapsed_sec + je->eta_sec);
365 eta_to_str(eta_str, je->eta_sec);
366 }
367
368 p += sprintf(p, "Jobs: %d (f=%d)", je->nr_running, je->files_open);
369 if (je->m_rate || je->t_rate) {
370 char *tr, *mr;
371
372 mr = num2str(je->m_rate, 4, 0, i2p);
373 tr = num2str(je->t_rate, 4, 0, i2p);
374 p += sprintf(p, ", CR=%s/%s KB/s", tr, mr);
375 free(tr);
376 free(mr);
377 } else if (je->m_iops || je->t_iops)
378 p += sprintf(p, ", CR=%d/%d IOPS", je->t_iops, je->m_iops);
379 if (je->eta_sec != INT_MAX && je->nr_running) {
380 char perc_str[32];
381 char *iops_str[2];
382 char *rate_str[2];
383 int l;
384
385 if ((!je->eta_sec && !eta_good) || je->nr_ramp == je->nr_running)
386 strcpy(perc_str, "-.-% done");
387 else {
388 eta_good = 1;
389 perc *= 100.0;
390 sprintf(perc_str, "%3.1f%% done", perc);
391 }
392
393 rate_str[0] = num2str(je->rate[0], 5, 10, i2p);
394 rate_str[1] = num2str(je->rate[1], 5, 10, i2p);
395
396 iops_str[0] = num2str(je->iops[0], 4, 1, 0);
397 iops_str[1] = num2str(je->iops[1], 4, 1, 0);
398
399 l = sprintf(p, ": [%s] [%s] [%s/%s /s] [%s/%s iops] [eta %s]",
400 je->run_str, perc_str, rate_str[0],
401 rate_str[1], iops_str[0], iops_str[1], eta_str);
402 p += l;
403 if (l >= 0 && l < linelen_last)
404 p += sprintf(p, "%*s", linelen_last - l, "");
405 linelen_last = l;
406
407 free(rate_str[0]);
408 free(rate_str[1]);
409 free(iops_str[0]);
410 free(iops_str[1]);
411 }
412 p += sprintf(p, "\r");
413
414 printf("%s", output);
415 fflush(stdout);
416}
417
418void print_thread_status(void)
419{
420 struct jobs_eta *je;
421 size_t size;
422
423 if (!thread_number)
424 return;
425
426 size = sizeof(*je) + thread_number * sizeof(char) + 1;
427 je = malloc(size);
428 memset(je, 0, size);
429
430 if (calc_thread_status(je, 0))
431 display_thread_status(je);
432
433 free(je);
434}
435
436void print_status_init(int thr_number)
437{
438 run_str[thr_number] = 'P';
439}