KolibriOS/umka
10
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[umka_os] Introduce dedicated I/O thread

Browse Source 
Not as nice as io_uring but portable.
This commit is contained in:
Ivan Baravy 2023-02-02 23:52:35 +00:00
parent 12442c72a7
commit d0387f7d53
8 changed files with 185 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
io_async.h
View File

@ -1,27 +0,0 @@
/*
SPDX-License-Identifier: GPL-2.0-or-later
UMKa - User-Mode KolibriOS developer tools
io_async - input/output platform specific code
Copyright (C) 2023 Ivan Baravy <dunkaist@gmail.com>
*/
#ifndef IO_ASYNC_H_INCLUDED
#define IO_ASYNC_H_INCLUDED
#include <stddef.h>
void *
io_async_init();
void
io_async_close(void *arg);
ssize_t
io_async_read(int fd, void *buf, size_t count, void *arg);
ssize_t
io_async_write(int fd, const void *buf, size_t count, void *arg);
#endif // IO_ASYNC_H_INCLUDED

212
linux/io_async.c
View File

@ -1,212 +0,0 @@
/*
SPDX-License-Identifier: GPL-2.0-or-later
UMKa - User-Mode KolibriOS developer tools
io_async - input/output platform specific code
Copyright (C) 2023 Ivan Baravy <dunkaist@gmail.com>
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <inttypes.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include <linux/io_uring.h>
#include "../umka.h"
#define QUEUE_DEPTH 1
#define read_barrier() __asm__ __volatile__("":::"memory")
#define write_barrier() __asm__ __volatile__("":::"memory")
#define MAX(x, y) ((x) >= (y) ? (x) : (y))
struct io_uring_queue {
int fd;
void *base;
size_t base_mmap_len;
struct io_uring_params p;
uint32_t *array;
struct io_uring_sqe *sqes;
size_t sqes_mmap_len;
uint32_t *sq_head, *sq_tail;
uint32_t sq_mask;
struct io_uring_cqe *cqes;
uint32_t *cq_head, *cq_tail;
uint32_t cq_mask;
};
static int
io_uring_setup(unsigned entries, struct io_uring_params *p) {
return (int) syscall(__NR_io_uring_setup, entries, p);
}
static int
io_uring_enter(int ring_fd, unsigned int to_submit, unsigned int min_complete,
unsigned int flags) {
return (int) syscall(__NR_io_uring_enter, ring_fd, to_submit, min_complete,
flags, NULL, 0);
}
/*
static void
dump_scht(char *prefix, struct io_uring_queue *q) {
fprintf(stderr, "# async %s: %p %p\n", prefix, (void*)q, q->base);
size_t head, tail, mask;
read_barrier();
mask = q->sq_mask;
mask = *(uint32_t*)(((uintptr_t)q->base) + q->p.sq_off.ring_mask);
head = *q->sq_head & mask;
tail = *q->sq_tail & mask;
fprintf(stderr, "######### %s ######### sq %u:%u 0x%x\n", prefix, head, tail, mask);
mask = q->cq_mask;
mask = *(uint32_t*)(((uintptr_t)q->base) + q->p.cq_off.ring_mask);
head = *q->cq_head & mask;
tail = *q->cq_tail & mask;
fprintf(stderr, "######### %s ######### cq %u:%u 0x%x\n", prefix, head, tail, mask);
}
*/
int
cq_has_data(struct io_uring_queue *q) {
size_t head, tail;
read_barrier();
head = *q->cq_head & q->cq_mask;
tail = *q->cq_tail & q->cq_mask;
return head != tail;
}
static void
build_op_read(struct io_uring_sqe *sqe, int fd, void *data, size_t size,
off_t offset) {
sqe->fd = fd;
sqe->flags = 0;
sqe->opcode = IORING_OP_READ;
sqe->addr = (uintptr_t)data;
sqe->len = size;
sqe->off = offset;
sqe->user_data = 0;
}
static int
read_from_cq(struct io_uring_queue *q) {
size_t head, tail;
struct io_uring_cqe *cqe;
do {
read_barrier();
head = *q->cq_head & q->cq_mask;
tail = *q->cq_tail & q->cq_mask;
if (head == tail)
break;
/* Get the entry */
cqe = q->cqes + head;
if (cqe->res < 0) {
fprintf(stderr, "Read error: %s\n", strerror(abs(cqe->res)));
}
(*q->cq_head)++;
} while (head == tail);
write_barrier();
return cqe->res;
}
struct io_uring_queue *
io_async_init() {
struct io_uring_queue *q = calloc(1, sizeof(struct io_uring_queue));
q->fd = io_uring_setup(QUEUE_DEPTH, &q->p);
if (q->fd < 0) {
perror("io_uring_setup");
return NULL;
}
size_t sring_size = q->p.sq_off.array + q->p.sq_entries * sizeof(unsigned);
size_t cring_size = q->p.cq_off.cqes
+ q->p.cq_entries * sizeof(struct io_uring_cqe);
size_t rings_size = MAX(sring_size, cring_size);
q->base_mmap_len = rings_size;
if (!(q->p.features & IORING_FEAT_SINGLE_MMAP)) {
fprintf(stderr, "[io.uring] Your kernel doesn't support"
" IORING_FEAT_SINGLE_MMAP, upgrade to Linux 5.4\n");
return NULL;
}
q->base = mmap(0, rings_size, PROT_READ | PROT_WRITE, MAP_SHARED
| MAP_POPULATE, q->fd, IORING_OFF_SQ_RING);
if (q->base == MAP_FAILED) {
perror("[io.uring] Can't mmap io_uring rings");
return NULL;
}
q->sqes_mmap_len = q->p.sq_entries * sizeof(struct io_uring_sqe);
q->sqes = mmap(0, q->sqes_mmap_len, PROT_READ | PROT_WRITE, MAP_SHARED
| MAP_POPULATE, q->fd, IORING_OFF_SQES);
if (q->sqes == MAP_FAILED) {
perror("[io.uring] Can't mmap io_uring SQEs");
return NULL;
}
q->cqes = (struct io_uring_cqe*)((uintptr_t)q->base + q->p.cq_off.cqes);
q->array = (uint32_t*)((uintptr_t)q->base + q->p.sq_off.array);
q->sq_head = (uint32_t*)((uintptr_t)q->base + q->p.sq_off.head);
q->sq_tail = (uint32_t*)((uintptr_t)q->base + q->p.sq_off.tail);
q->sq_mask = *(uint32_t*)((uintptr_t)q->base + q->p.sq_off.ring_mask);
q->cq_head = (uint32_t*)((uintptr_t)q->base + q->p.cq_off.head);
q->cq_tail = (uint32_t*)((uintptr_t)q->base + q->p.cq_off.tail);
q->cq_mask = *(uint32_t*)((uintptr_t)q->base + q->p.cq_off.ring_mask);
return q;
}
void
io_async_close(void *arg) {
struct io_uring_queue *q = arg;
munmap(q->base, q->base_mmap_len);
munmap(q->sqes, q->sqes_mmap_len);
close(q->fd);
free(q);
}
static uint32_t
io_uring_wait_test() {
appdata_t *app;
__asm__ __volatile__ ("":"=b"(app)::);
struct io_uring_queue *q = app->wait_param;
int done = cq_has_data(q);
return done;
}
ssize_t
io_async_read(int fd, void *buf, size_t count, void *arg) {
struct io_uring_queue *q = arg;
read_barrier();
size_t tail = *q->sq_tail & q->sq_mask;
struct io_uring_sqe *sqe = q->sqes + tail;
build_op_read(sqe, fd, buf, count, -1);
q->array[tail] = tail;
(*q->sq_tail)++;
int ret = io_uring_enter(q->fd, 1, 0, IORING_ENTER_GETEVENTS);
if(ret < 0) {
perror("io_uring_enter");
return 1;
}
kos_wait_events(io_uring_wait_test, q);
ssize_t res = read_from_cq(q);
return res;
}
ssize_t
io_async_write(int fd, const void *buf, size_t count, void *arg) {
(void)fd;
(void)buf;
(void)count;
(void)arg;
return -1;
}

12
makefile
View File

@ -58,19 +58,18 @@ test: umka_shell
umka_shell: umka_shell.o umka.o shell.o trace.o trace_lbr.o vdisk.o \
vdisk/raw.o vdisk/qcow2.o vdisk/miniz/miniz.a vnet.o \
$(HOST)/vnet/tap.o vnet/file.o lodepng.o $(HOST)/pci.o \
$(HOST)/thread.o umkaio.o $(HOST)/io_async.o umkart.o optparse32.o \
bestline32.o
$(HOST)/thread.o umkaio.o umkart.o optparse32.o bestline32.o
$(CC) $(LDFLAGS_32) $^ -o $@ -T umka.ld
umka_fuse: umka_fuse.o umka.o trace.o trace_lbr.o vdisk.o vdisk/raw.o \
vdisk/qcow2.o vdisk/miniz/miniz.a $(HOST)/pci.o $(HOST)/thread.o \
umkaio.o $(HOST)/io_async.o
umkaio.o
$(CC) $(LDFLAGS_32) $^ -o $@ `pkg-config fuse3 --libs` -T umka.ld
umka_os: umka_os.o umka.o shell.o lodepng.o vdisk.o vdisk/raw.o vdisk/qcow2.o \
vdisk/miniz/miniz.a vnet.o $(HOST)/vnet/tap.o vnet/file.o trace.o \
trace_lbr.o $(HOST)/pci.o $(HOST)/thread.o umkaio.o \
$(HOST)/io_async.o umkart.o bestline32.o optparse32.o
trace_lbr.o $(HOST)/pci.o $(HOST)/thread.o umkaio.o umkart.o \
bestline32.o optparse32.o
$(CC) $(LDFLAGS_32) `sdl2-config --libs` $^ -o $@ -T umka.ld
umka_gen_devices_dat: umka_gen_devices_dat.o umka.o $(HOST)/pci.o \
@ -86,9 +85,6 @@ shell.o: shell.c lodepng.h
umkaio.o: umkaio.c umkaio.h
$(CC) $(CFLAGS_32) -D_DEFAULT_SOURCE -c $< -o $@
$(HOST)/io_async.o: $(HOST)/io_async.c io_async.h
$(CC) $(CFLAGS_32) -D_DEFAULT_SOURCE -c $< -o $@
$(HOST)/thread.o: $(HOST)/thread.c
$(CC) $(CFLAGS_32) -c $< -o $@

19
shell.c
View File

@ -75,6 +75,11 @@ umka_run_cmd_sync(struct shell_ctx *ctx) {
c->hscroll);
break;
}
case UMKA_CMD_SYS_LFN: {
struct cmd_sys_lfn *c = &cmd->arg.sys_lfn;
umka_sys_lfn(c->bufptr, c->r, c->f70or80);
break;
}
default:
fprintf(ctx->fout, "[!] unknown command: %u\n", cmd->type);
break;
@ -2353,9 +2358,21 @@ ls_all(struct shell_ctx *ctx, f7080s1arg_t *fX0, f70or80_t f70or80) {
size_t bdfe_len = (fX0->encoding == CP866) ? BDFE_LEN_CP866 :
BDFE_LEN_UNICODE;
while (true) {
struct umka_cmd *cmd = umka_cmd_buf;
struct cmd_sys_lfn *c = &cmd->arg.sys_lfn;
cmd->type = UMKA_CMD_SYS_LFN;
c->f70or80 = f70or80;
c->bufptr = fX0;
c->r = &r;
atomic_store_explicit(&cmd->status, UMKA_CMD_STATUS_READY,
memory_order_release);
COVERAGE_ON();
umka_sys_lfn(fX0, &r, f70or80);
// umka_sys_lfn(fX0, &r, f70or80);
umka_run_cmd(ctx);
COVERAGE_OFF();
atomic_store_explicit(&cmd->status, UMKA_CMD_STATUS_EMPTY,
memory_order_release);
print_f70_status(ctx, &r, 1);
assert((r.status == ERROR_SUCCESS && r.count == fX0->size)
|| (r.status == ERROR_END_OF_FILE && r.count < fX0->size));

13
umka.h
View File

@ -2401,6 +2401,7 @@ enum {
UMKA_CMD_SET_MOUSE_DATA,
UMKA_CMD_SYS_PROCESS_INFO,
UMKA_CMD_SYS_GET_MOUSE_POS_SCREEN,
UMKA_CMD_SYS_LFN,
};
enum {
@ -2417,6 +2418,16 @@ struct cmd_set_mouse_data {
int32_t hscroll;
};
struct cmd_sys_lfn {
f70or80_t f70or80;
f7080s1arg_t *bufptr;
f7080ret_t *r;
};
struct cmd_ret_sys_lfn {
f7080ret_t status;
};
struct cmd_sys_process_info {
int32_t pid;
void *param;
@ -2431,9 +2442,11 @@ struct umka_cmd {
uint32_t type;
union {
struct cmd_set_mouse_data set_mouse_data;
struct cmd_sys_lfn sys_lfn;
} arg;
union {
struct cmd_ret_sys_get_mouse_pos_screen sys_get_mouse_pos_screen;
struct cmd_ret_sys_lfn sys_lfn;
} ret;
};

2
umka_shell.c
View File

@ -35,7 +35,7 @@ struct umka_shell_ctx *
umka_shell_init(int reproducible, FILE *fin, FILE *fout) {
struct umka_shell_ctx *ctx = malloc(sizeof(struct umka_shell_ctx));
ctx->umka = umka_init();
ctx->io = io_init(&ctx->umka->running);
ctx->io = io_init(NULL);
ctx->shell = shell_init(reproducible, history_filename, ctx->umka, ctx->io,
fin, fout, &ctx->umka->running);
return ctx;

106
umkaio.c
View File

@ -7,35 +7,127 @@
Copyright (C) 2023 Ivan Baravy <dunkaist@gmail.com>
*/
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <inttypes.h>
#include "umka.h"
#include "umkaio.h"
#include "io_async.h"
#define IOT_QUEUE_DEPTH 1
struct iot_cmd iot_cmd_buf[IOT_QUEUE_DEPTH];
static void *
thread_io(void *arg) {
(void)arg;
for (size_t i = 0; i < IOT_QUEUE_DEPTH; i++) {
iot_cmd_buf[i].status = UMKA_CMD_STATUS_EMPTY;
iot_cmd_buf[i].type = 0;
pthread_cond_init(&iot_cmd_buf[i].iot_cond, NULL);
pthread_mutex_init(&iot_cmd_buf[i].iot_mutex, NULL);
pthread_mutex_lock(&iot_cmd_buf[i].iot_mutex);
pthread_mutex_init(&iot_cmd_buf[i].mutex, NULL);
}
struct iot_cmd *cmd = iot_cmd_buf;
ssize_t ret;
while (1) {
pthread_cond_wait(&cmd->iot_cond, &cmd->iot_mutex);
// status must be ready
switch (cmd->type) {
case IOT_CMD_TYPE_READ:
ret = read(cmd->read.arg.fd, cmd->read.arg.buf, cmd->read.arg.count);
atomic_store_explicit(&cmd->read.ret.val, ret, memory_order_release);
break;
case IOT_CMD_TYPE_WRITE:
cmd->read.ret.val = write(cmd->read.arg.fd, cmd->read.arg.buf,
cmd->read.arg.count);
break;
default:
break;
}
atomic_store_explicit(&cmd->status, UMKA_CMD_STATUS_DONE, memory_order_release);
}
return NULL;
}
static uint32_t
io_async_submit_wait_test() {
// appdata_t *app;
// __asm__ __volatile__ ("":"=b"(app)::);
// struct io_uring_queue *q = app->wait_param;
int done = pthread_mutex_trylock(&iot_cmd_buf[0].mutex);
return done;
}
static uint32_t
io_async_complete_wait_test() {
// appdata_t *app;
// __asm__ __volatile__ ("":"=b"(app)::);
// struct io_uring_queue *q = app->wait_param;
int status = atomic_load_explicit(&iot_cmd_buf[0].status, memory_order_acquire);
return status == UMKA_CMD_STATUS_DONE;
}
ssize_t
io_async_read(int fd, void *buf, size_t count, void *arg) {
(void)arg;
kos_wait_events(io_async_submit_wait_test, NULL);
// status must be empty
struct iot_cmd *cmd = iot_cmd_buf;
cmd->read.arg.fd = fd;
cmd->read.arg.buf = buf;
cmd->read.arg.count = count;
atomic_store_explicit(&cmd->status, UMKA_CMD_STATUS_READY, memory_order_release);
pthread_cond_signal(&cmd->iot_cond);
kos_wait_events(io_async_complete_wait_test, NULL);
ssize_t res = atomic_load_explicit(&cmd->read.ret.val, memory_order_acquire);
atomic_store_explicit(&cmd->status, UMKA_CMD_STATUS_EMPTY, memory_order_release);
pthread_mutex_unlock(&cmd->mutex);
return res;
}
ssize_t
io_async_write(int fd, const void *buf, size_t count, void *arg) {
(void)fd;
(void)buf;
(void)count;
(void)arg;
return -1;
}
struct umka_io *
io_init(int *running) {
struct umka_io *io = malloc(sizeof(struct umka_io));
io->running = running;
io->async = io_async_init();
if (running) {
pthread_create(&io->iot, NULL, thread_io, NULL);
}
return io;
}
void
io_close(struct umka_io *io) {
io_async_close(io->async);
free(io);
}
ssize_t
io_read(int fd, void *buf, size_t count, struct umka_io *io) {
ssize_t res;
if (!*io->running) {
if (!io->running || !*io->running) {
res = read(fd, buf, count);
} else {
res = io_async_read(fd, buf, count, io->async);
res = io_async_read(fd, buf, count, NULL);
}
return res;
}
@ -43,10 +135,10 @@ io_read(int fd, void *buf, size_t count, struct umka_io *io) {
ssize_t
io_write(int fd, const void *buf, size_t count, struct umka_io *io) {
ssize_t res;
if (!*io->running) {
if (!io->running || !*io->running) {
res = write(fd, buf, count);
} else {
res = io_async_write(fd, buf, count, io->async);
res = io_async_write(fd, buf, count, NULL);
}
return res;
}

52
umkaio.h
View File

@ -10,11 +10,59 @@
#ifndef UMKAIO_H_INCLUDED
#define UMKAIO_H_INCLUDED
#include <unistd.h>
#include <stddef.h>
#include <pthread.h>
struct umka_io {
const int *running;
void *async; // platform specific
pthread_t iot;
};
struct iot_cmd_read_arg {
int fd;
void *buf;
size_t count;
};
struct iot_cmd_read_ret {
ssize_t val;
};
union iot_cmd_read {
struct iot_cmd_read_arg arg;
struct iot_cmd_read_ret ret;
};
struct iot_cmd_write_arg {
int fd;
void *buf;
size_t count;
};
struct iot_cmd_write_ret {
ssize_t val;
};
union iot_cmd_write {
struct iot_cmd_write_arg arg;
struct iot_cmd_write_ret ret;
};
enum {
IOT_CMD_TYPE_READ,
IOT_CMD_TYPE_WRITE,
};
struct iot_cmd {
pthread_cond_t iot_cond;
pthread_mutex_t iot_mutex;
pthread_mutex_t mutex;
int status;
int type;
union {
union iot_cmd_read read;
union iot_cmd_write write;
};
};
struct umka_io *