Don't use seeks on stdin for job file parsing
[fio.git] / engines / syslet-rw.c
1 /*
2  * syslet engine
3  *
4  * IO engine that does regular pread(2)/pwrite(2) to transfer data, but
5  * with syslets to make the execution async.
6  *
7  */
8 #include <stdio.h>
9 #include <stdlib.h>
10 #include <unistd.h>
11 #include <errno.h>
12 #include <assert.h>
13 #include <asm/unistd.h>
14
15 #include "../fio.h"
16
17 #ifdef FIO_HAVE_SYSLET
18
19 #ifdef __NR_pread64
20 #define __NR_fio_pread  __NR_pread64
21 #define __NR_fio_pwrite __NR_pwrite64
22 #else
23 #define __NR_fio_pread  __NR_pread
24 #define __NR_fio_pwrite __NR_pwrite
25 #endif
26
27 struct syslet_data {
28         struct io_u **events;
29         unsigned int nr_events;
30         
31         struct async_head_user ahu;
32         struct syslet_uatom **ring;
33
34         struct syslet_uatom *head, *tail;
35 };
36
37 static void fio_syslet_complete_atom(struct thread_data *td,
38                                      struct syslet_uatom *atom)
39 {
40         struct syslet_data *sd = td->io_ops->data;
41         struct syslet_uatom *last;
42         struct io_u *io_u;
43
44         /*
45          * complete from the beginning of the sequence up to (and
46          * including) this atom
47          */
48         last = atom;
49         io_u = (struct io_u *)atom->private;
50         atom = io_u->req.head;
51
52         /*
53          * now complete in right order
54          */
55         do {
56                 long ret;
57
58                 io_u = (struct io_u *)atom->private;
59                 ret = *(long *)atom->ret_ptr;
60                 if (ret >= 0)
61                         io_u->resid = io_u->xfer_buflen - ret;
62                 else if (ret < 0)
63                         io_u->error = ret;
64
65                 assert(sd->nr_events < td->o.iodepth);
66                 sd->events[sd->nr_events++] = io_u;
67
68                 if (atom == last)
69                         break;
70
71                 atom = (struct syslet_uatom *)atom->next;
72         } while (1);
73
74         assert(!last->next);
75 }
76
77 /*
78  * Inspect the ring to see if we have completed events
79  */
80 static void fio_syslet_complete(struct thread_data *td)
81 {
82         struct syslet_data *sd = td->io_ops->data;
83
84         do {
85                 struct syslet_uatom *atom;
86
87                 atom = sd->ring[sd->ahu.user_ring_idx];
88                 if (!atom)
89                         break;
90
91                 sd->ring[sd->ahu.user_ring_idx] = NULL;
92                 if (++sd->ahu.user_ring_idx == td->o.iodepth)
93                         sd->ahu.user_ring_idx = 0;
94
95                 fio_syslet_complete_atom(td, atom);
96         } while (1);
97 }
98
99 static int fio_syslet_getevents(struct thread_data *td, int min,
100                                 int fio_unused max,
101                                 struct timespec fio_unused *t)
102 {
103         struct syslet_data *sd = td->io_ops->data;
104         long ret;
105
106         do {
107                 fio_syslet_complete(td);
108
109                 /*
110                  * do we have enough immediate completions?
111                  */
112                 if (sd->nr_events >= (unsigned int) min)
113                         break;
114
115                 /*
116                  * OK, we need to wait for some events...
117                  */
118                 ret = async_wait(1, sd->ahu.user_ring_idx, &sd->ahu);
119                 if (ret < 0)
120                         return -errno;
121         } while (1);
122
123         ret = sd->nr_events;
124         sd->nr_events = 0;
125         return ret;
126 }
127
128 static struct io_u *fio_syslet_event(struct thread_data *td, int event)
129 {
130         struct syslet_data *sd = td->io_ops->data;
131
132         return sd->events[event];
133 }
134
135 static void init_atom(struct syslet_uatom *atom, int nr, void *arg0,
136                       void *arg1, void *arg2, void *arg3, void *ret_ptr,
137                       unsigned long flags, void *priv)
138 {
139         atom->flags = flags;
140         atom->nr = nr;
141         atom->ret_ptr = (uint64_t)ret_ptr;
142         atom->next = 0;
143         atom->arg_ptr[0] = (uint64_t)arg0;
144         atom->arg_ptr[1] = (uint64_t)arg1;
145         atom->arg_ptr[2] = (uint64_t)arg2;
146         atom->arg_ptr[3] = (uint64_t)arg3;
147         atom->arg_ptr[4] = 0;
148         atom->arg_ptr[5] = 0;
149         atom->private = (uint64_t)priv;
150 }
151
152 /*
153  * Use seek atom for sync
154  */
155 static void fio_syslet_prep_sync(struct io_u *io_u, struct fio_file *f)
156 {
157         init_atom(&io_u->req.atom, __NR_fsync, &f->fd, NULL, NULL, NULL,
158                   &io_u->req.ret, 0, io_u);
159 }
160
161 static void fio_syslet_prep_rw(struct io_u *io_u, struct fio_file *f)
162 {
163         int nr;
164
165         /*
166          * prepare rw
167          */
168         if (io_u->ddir == DDIR_READ)
169                 nr = __NR_fio_pread;
170         else
171                 nr = __NR_fio_pwrite;
172
173         init_atom(&io_u->req.atom, nr, &f->fd, &io_u->xfer_buf,
174                   &io_u->xfer_buflen, &io_u->offset, &io_u->req.ret, 0, io_u);
175 }
176
177 static int fio_syslet_prep(struct thread_data fio_unused *td, struct io_u *io_u)
178 {
179         struct fio_file *f = io_u->file;
180
181         if (io_u->ddir == DDIR_SYNC)
182                 fio_syslet_prep_sync(io_u, f);
183         else
184                 fio_syslet_prep_rw(io_u, f);
185
186         return 0;
187 }
188
189 static void cachemiss_thread_start(void)
190 {
191         while (1)
192                 async_thread(NULL, NULL);
193 }
194
195 #define THREAD_STACK_SIZE (16384)
196
197 static unsigned long thread_stack_alloc()
198 {
199         return (unsigned long) malloc(THREAD_STACK_SIZE) + THREAD_STACK_SIZE;
200 }
201
202 static void fio_syslet_queued(struct thread_data *td, struct syslet_data *sd)
203 {
204         struct syslet_uatom *atom;
205         struct timeval now;
206
207         fio_gettime(&now, NULL);
208
209         atom = sd->head;
210         while (atom) {
211                 struct io_u *io_u = (struct io_u *)atom->private;
212
213                 memcpy(&io_u->issue_time, &now, sizeof(now));
214                 io_u_queued(td, io_u);
215                 atom = (struct syslet_uatom *)atom->next;
216         }
217 }
218
219 static int fio_syslet_commit(struct thread_data *td)
220 {
221         struct syslet_data *sd = td->io_ops->data;
222         struct syslet_uatom *done;
223
224         if (!sd->head)
225                 return 0;
226
227         assert(!sd->tail->next);
228
229         if (!sd->ahu.new_thread_stack)
230                 sd->ahu.new_thread_stack = thread_stack_alloc();
231
232         fio_syslet_queued(td, sd);
233
234         /*
235          * On sync completion, the atom is returned. So on NULL return
236          * it's queued asynchronously.
237          */
238         done = async_exec(sd->head, &sd->ahu);
239
240         if (done == (void *) -1) {
241                 log_err("fio: syslets don't appear to work\n");
242                 return -1;
243         }
244
245         sd->head = sd->tail = NULL;
246
247         if (done)
248                 fio_syslet_complete_atom(td, done);
249
250         return 0;
251 }
252
253 static int fio_syslet_queue(struct thread_data *td, struct io_u *io_u)
254 {
255         struct syslet_data *sd = td->io_ops->data;
256
257         fio_ro_check(td, io_u);
258
259         if (sd->tail) {
260                 sd->tail->next = (uint64_t)&io_u->req.atom;
261                 sd->tail = &io_u->req.atom;
262         } else
263                 sd->head = sd->tail = (struct syslet_uatom *)&io_u->req.atom;
264
265         io_u->req.head = sd->head;
266         return FIO_Q_QUEUED;
267 }
268
269 static int async_head_init(struct syslet_data *sd, unsigned int depth)
270 {
271         unsigned long ring_size;
272
273         memset(&sd->ahu, 0, sizeof(struct async_head_user));
274
275         ring_size = sizeof(struct syslet_uatom *) * depth;
276         sd->ring = malloc(ring_size);
277         memset(sd->ring, 0, ring_size);
278
279         sd->ahu.user_ring_idx = 0;
280         sd->ahu.completion_ring_ptr = (uint64_t)sd->ring;
281         sd->ahu.ring_size_bytes = ring_size;
282         sd->ahu.head_stack = thread_stack_alloc();
283         sd->ahu.head_ip = (uint64_t)cachemiss_thread_start;
284         sd->ahu.new_thread_ip = (uint64_t)cachemiss_thread_start;
285         sd->ahu.new_thread_stack = thread_stack_alloc();
286
287         return 0;
288 }
289
290 static void async_head_exit(struct syslet_data *sd)
291 {
292         free(sd->ring);
293 }
294
295 static int check_syslet_support(struct syslet_data *sd)
296 {
297         struct syslet_uatom atom;
298         void *ret;
299
300         init_atom(&atom, __NR_getpid, NULL, NULL, NULL, NULL, NULL, 0, NULL);
301         ret = async_exec(&atom, &sd->ahu);
302         if (ret == (void *) -1)
303                 return 1;
304
305         return 0;
306 }
307
308 static void fio_syslet_cleanup(struct thread_data *td)
309 {
310         struct syslet_data *sd = td->io_ops->data;
311
312         if (sd) {
313                 async_head_exit(sd);
314                 free(sd->events);
315                 free(sd);
316                 td->io_ops->data = NULL;
317         }
318 }
319
320 static int fio_syslet_init(struct thread_data *td)
321 {
322         struct syslet_data *sd;
323
324         sd = malloc(sizeof(*sd));
325         memset(sd, 0, sizeof(*sd));
326         sd->events = malloc(sizeof(struct io_u *) * td->o.iodepth);
327         memset(sd->events, 0, sizeof(struct io_u *) * td->o.iodepth);
328
329         /*
330          * This will handily fail for kernels where syslet isn't available
331          */
332         if (async_head_init(sd, td->o.iodepth)) {
333                 free(sd->events);
334                 free(sd);
335                 return 1;
336         }
337
338         if (check_syslet_support(sd)) {
339                 log_err("fio: syslets do not appear to work\n");
340                 free(sd->events);
341                 free(sd);
342                 return 1;
343         }
344
345         td->io_ops->data = sd;
346         return 0;
347 }
348
349 static struct ioengine_ops ioengine = {
350         .name           = "syslet-rw",
351         .version        = FIO_IOOPS_VERSION,
352         .init           = fio_syslet_init,
353         .prep           = fio_syslet_prep,
354         .queue          = fio_syslet_queue,
355         .commit         = fio_syslet_commit,
356         .getevents      = fio_syslet_getevents,
357         .event          = fio_syslet_event,
358         .cleanup        = fio_syslet_cleanup,
359         .open_file      = generic_open_file,
360         .close_file     = generic_close_file,
361 };
362
363 #else /* FIO_HAVE_SYSLET */
364
365 /*
366  * When we have a proper configure system in place, we simply wont build
367  * and install this io engine. For now install a crippled version that
368  * just complains and fails to load.
369  */
370 static int fio_syslet_init(struct thread_data fio_unused *td)
371 {
372         fprintf(stderr, "fio: syslet not available\n");
373         return 1;
374 }
375
376 static struct ioengine_ops ioengine = {
377         .name           = "syslet-rw",
378         .version        = FIO_IOOPS_VERSION,
379         .init           = fio_syslet_init,
380 };
381
382 #endif /* FIO_HAVE_SYSLET */
383
384 static void fio_init fio_syslet_register(void)
385 {
386         register_ioengine(&ioengine);
387 }
388
389 static void fio_exit fio_syslet_unregister(void)
390 {
391         unregister_ioengine(&ioengine);
392 }