diff options
Diffstat (limited to 'seqreader.c')
| -rw-r--r-- | seqreader.c | 130 |
1 files changed, 123 insertions, 7 deletions
diff --git a/seqreader.c b/seqreader.c index 359f38e..ba4db51 100644 --- a/seqreader.c +++ b/seqreader.c @@ -2,6 +2,7 @@ #include <stdio.h> #include <stdlib.h> +#include <stdint.h> #include <sys/types.h> #include <sys/stat.h> #include <sys/time.h> @@ -11,16 +12,34 @@ #include <string.h> #include <errno.h> +#include <libaio.h> + #define FASTWRITER_SYNCIO_ALIGN 512 #define SYNC_MODE -#define BUFSIZE 2097152 +#define AIO_MODE 2 +#define EXTRA_BUFFERS 2 +#define WRITE_INTERVAL 1 + +#define RAID_STRIP_SIZE 256 +#define RAID_DISKS 8 +#define STRIPS_AT_ONCE 2 + +#ifdef AIO_MODE +# define SYNC_MODE +#endif /* AIO_MODE */ + #ifdef SYNC_MODE -# define BLOCK_SIZE 2097152 +# define BLOCK_SIZE (1024 * RAID_STRIP_SIZE * RAID_DISKS * STRIPS_AT_ONCE) #else /* SYNC_MODE */ # define BLOCK_SIZE 16384 #endif /* SYNC_MODE */ -#define WRITE_INTERVAL 1 + +#ifdef AIO_MODE +# define BUFSIZE (BLOCK_SIZE * (AIO_MODE + EXTRA_BUFFERS)) +#else /* AIO_MODE */ +# define BUFSIZE BLOCK_SIZE +#endif /* AIO_MODE */ int main(int argc, char *argv[]) { @@ -33,18 +52,34 @@ int main(int argc, char *argv[]) { size_t files = 0; size_t total_size = 0; size_t last_write = 0; + size_t last_size = 0; size_t skip; size_t run; + size_t ready; ssize_t res; size_t max_size = (size_t)-1; char *buffer;//[BUFSIZE]; long double mcoef = 1000000. / (1024 * 1024); int flags = O_RDONLY|O_NOATIME|O_LARGEFILE; +#ifdef AIO_MODE + int i; + size_t curio, schedio; + int done[AIO_MODE + EXTRA_BUFFERS]; + + io_context_t aio; + struct iocb io[AIO_MODE], *ioptr[AIO_MODE]; + + int events; + struct io_event ev[AIO_MODE]; +#endif /* AIO_MODE */ + #ifdef SYNC_MODE flags |= O_DIRECT; #endif - + + printf("Used buffer: %i MB, Block: %i KB\n", BUFSIZE / 1024 / 1024, BLOCK_SIZE/1024); + posix_memalign((void**)&buffer, FASTWRITER_SYNCIO_ALIGN, BUFSIZE); if (argc < 2) { @@ -63,13 +98,87 @@ int main(int argc, char *argv[]) { printf("Unable to open device %s\n", argv[1]); exit(1); } - + size_t size = BLOCK_SIZE; + +#ifdef AIO_MODE + memset(done, 0, sizeof(done)); + memset(&aio, 0, sizeof(aio)); + io_queue_init(AIO_MODE, &aio); + for (i = 0; i < AIO_MODE; i++) { + ioptr[i] = &io[i]; + memset(ioptr[i], 0, sizeof(struct iocb)); + io_prep_pread(ioptr[i], fd, buffer + i * BLOCK_SIZE, BLOCK_SIZE, i * BLOCK_SIZE); + io_set_callback(ioptr[i], (void*)(uintptr_t)i); + } + + curio = 0; + schedio = AIO_MODE; + events = 0; +#endif /* AIO_MODE */ gettimeofday(&start, NULL); - + +#ifdef AIO_MODE + err = io_submit(aio, AIO_MODE, ioptr); + if (err != AIO_MODE) { + printf("io_submit returned %i\n", err); + perror("Failed to submit initial AIO jobs"); + } +#endif /* AIO_MODE */ + +#ifdef AIO_MODE + ready = 0; + while (1) { + if (!done[curio%(AIO_MODE + EXTRA_BUFFERS)]) { + err = io_getevents(aio, 1, AIO_MODE - events, &ev[events], NULL); + if (err < 0) perror("Error waiting for AIO\n"); + + if ((!ready)&&(err > 1)) { + printf("*** Multiple read requests (%i of %i) are finished simultaneously. It is either:\n", err, AIO_MODE); + printf(" Small buffer size (%i KB)\n", BLOCK_SIZE/1024); + printf(" More parallel AIOs (%i) than supported by kernel, try %i\n", AIO_MODE, AIO_MODE - err); + } + + for (i = 0; i < err; i++) { + struct io_event *ep = &ev[events + i]; + int doneio = (uintptr_t)ep->data; + if (ep->res2 || (ep->res != BLOCK_SIZE)) perror("Error in async IO"); + done[doneio%(AIO_MODE + EXTRA_BUFFERS)] = 1; +// printf("done (%i): %i\n", i, doneio); + } + + events += err; + + for (i = events - 1; (i >= 0)&&((schedio - curio) < (AIO_MODE + EXTRA_BUFFERS)); i--) { + struct iocb *newio = (struct iocb *)ev[i].obj; + memset(newio, 0, sizeof(struct iocb)); + io_prep_pread(newio, fd, buffer + (schedio % (AIO_MODE + EXTRA_BUFFERS)) * BLOCK_SIZE, BLOCK_SIZE, schedio * BLOCK_SIZE); + io_set_callback(newio, (void*)(uintptr_t)schedio); + err = io_submit(aio, 1, &newio); + if (err != 1) perror("Failed to submit AIO jobs"); + schedio++; + } + events = i + 1; + + if (events) { + printf("*** Unprocessed events (%i), probably not enough buffer space...\n", events); +// printf(" curio (%zu), schedio (%zu)\n", curio, schedio); + } + + ready = 1; + continue; + } + + done[curio%(AIO_MODE + EXTRA_BUFFERS)] = 0; + curio++; + + res = BLOCK_SIZE; +#else /* AIO_MODE */ res = read(fd, buffer, size); while (res > 0) { +#endif /* AIO_MODE */ + if (res != size) { printf("Incomplete read: %zu bytes read instead of %zu\n", res, size); exit(-1); @@ -79,8 +188,9 @@ int main(int argc, char *argv[]) { gettimeofday(&tv, NULL); us = (tv.tv_sec - start.tv_sec) * 1000000 + (tv.tv_usec - start.tv_usec); if ((us - last_write) > WRITE_INTERVAL * 1000000) { + printf("Reading: %s (%lu GB), Measured speed: %zu MB/s, Current speed: %zu MB/s\n", argv[0], total_size / 1024 / 1024 / 1024, (size_t)(mcoef * total_size / us), (size_t)(mcoef * (total_size - last_size) / (us - last_write))); last_write = us; - printf("Reading: %s (%lu GB), Measured speed: %zu MB/s\n", argv[0], total_size / 1024 / 1024 / 1024, (size_t)(mcoef * total_size / us)); + last_size = total_size; } if (total_size > max_size) { @@ -88,9 +198,15 @@ int main(int argc, char *argv[]) { break; } +#ifndef AIO_MODE res = read(fd, buffer, size); +#endif /* AIO_MODE */ } +#ifdef AIO_MODE + io_queue_release(aio); +#endif /* AIO_MODE */ + close(fd); if (res < 0) { |