view release on metacpan or  search on metacpan

lib/IO/Uring.xs  view on Meta::CPAN

	URING_CONSTANT(FSYNC_DATASYNC);

	URING_CONSTANT(RECVSEND_POLL_FIRST);

	URING_CONSTANT(TIMEOUT_ABS);
	URING_CONSTANT(TIMEOUT_BOOTTIME);
	URING_CONSTANT(TIMEOUT_REALTIME);
	URING_CONSTANT(TIMEOUT_ETIME_SUCCESS);
	URING_CONSTANT(TIMEOUT_MULTISHOT);

	CONSTANT(RENAME_EXCHANGE);
	CONSTANT(RENAME_NOREPLACE);

	CONSTANT(AT_SYMLINK_FOLLOW);
	CONSTANT(AT_REMOVEDIR);

	URING_CONSTANT(POLL_UPDATE_EVENTS);
	URING_CONSTANT(POLL_UPDATE_USER_DATA);
	URING_CONSTANT(POLL_ADD_MULTI);

	CONSTANT(P_PID);
	CONSTANT(P_PGID);
	CONSTANT(P_PIDFD);
	CONSTANT(P_ALL);
	CONSTANT(WEXITED);
	CONSTANT(WSTOPPED);
	CONSTANT(WCONTINUED);
	CONSTANT(WNOWAIT);

IO::Uring new(class, UV entries, ...)
CODE:
	RETVAL = safecalloc(1, sizeof(struct ring));
	struct io_uring_params params = {};
	params.flags = IORING_SETUP_SINGLE_ISSUER | IORING_SETUP_COOP_TASKRUN | IORING_SETUP_DEFER_TASKRUN | IORING_SETUP_SUBMIT_ALL;
	for (int current = 2; current + 1 < items; current += 2) {
		STRLEN key_length;
		const char* key = SvPV(ST(current), key_length);
		if (key_length == 11 && strEQ(key, "cqe_entries")) {
			params.flags |= IORING_SETUP_CQSIZE;
			params.cq_entries = SvIV(ST(current+1));
		} else if (key_length == 6 && strEQ(key, "sqpoll")) {
			params.flags |= IORING_SETUP_SQPOLL;
			params.sq_thread_idle = SvIV(ST(current + 1));
		} else
			warn("Unknown named argument '%s'", key);
	}
	int ret = io_uring_queue_init_params(entries, &RETVAL->uring, &params);

	if (ret) {
		safefree(RETVAL);
		Perl_croak(aTHX_ "Could not create ring: %s", strerror(-ret));
	}
OUTPUT:
	RETVAL


IV CLONE_SKIP(sv)


void run_once(IO::Uring self, unsigned min_events = 1, Time::Spec timeout = NULL, sigset_t* sigmask = NULL)
PPCODE:
	struct io_uring_cqe *cqe;
	int result = io_uring_submit_and_wait_timeout(&self->uring, &cqe, min_events, timeout, sigmask);

	if (result == -1 && errno == EINTR)
		PERL_ASYNC_CHECK();

	unsigned head;

	ENTER;
	SAVETMPS;
	EXTEND(SP, 2);
	io_uring_for_each_cqe(&self->uring, head, cqe) {
		struct callback* callback_data = (struct callback*)io_uring_cqe_get_data(cqe);
		io_uring_cqe_seen(&self->uring, cqe);
		if (callback_data->callback) {
			PUSHMARK(SP);
			mPUSHi(cqe->res);
			mPUSHu(cqe->flags);
			PUTBACK;
			call_sv(callback_data->callback, G_VOID);
			if (!(cqe->flags & IORING_CQE_F_MORE)) {
				SvREFCNT_dec(callback_data->callback);
				if (callback_data->arg1)
					SvREFCNT_dec(callback_data->arg1);
				if (callback_data->arg2)
					SvREFCNT_dec(callback_data->arg2);
				Safefree(callback_data);
			}
			SPAGAIN;
			FREETMPS;
		}
		else if (!(cqe->flags & IORING_CQE_F_MORE)) {
			if (callback_data->arg1)
				SvREFCNT_dec(callback_data->arg1);
			if (callback_data->arg2)
				SvREFCNT_dec(callback_data->arg2);
			Safefree(callback_data);
		}
	}
	LEAVE;
	if (result >= 0)
		mPUSHi(result);
	else {
		errno = -result;
		PUSHs(&PL_sv_undef);
	}


SV* submit(IO::Uring self)
CODE:
	int result = io_uring_submit(&self->uring);
	if (result >= 0)
		RETVAL = newSViv(result);
	else {
		errno = -result;
		RETVAL = &PL_sv_undef;
	}
OUTPUT:
	RETVAL