danog/php-tokio
1
0
Fork 0
mirror of https://github.com/danog/php-tokio.git synced 2024-11-26 12:24:58 +01:00
Code Issues Packages Projects Releases Wiki Activity

Run cargo fmt

Browse Source
This commit is contained in:
Daniil Gentili 2023-08-29 20:50:06 +02:00
parent e063efcdf5
commit 1c0c2db96f
Signed by: danog
GPG Key ID: 8C1BE3B34B230CA7
4 changed files with 74 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
examples/reqwest/src/lib.rs
View File

@ -1,24 +1,19 @@
use php_tokio::{EventLoop, php_async_impl};
use nicelocal_ext_php_rs::prelude::*;
use php_tokio::{php_async_impl, EventLoop};
#[php_class]
struct Client {
}
struct Client {}
#[php_async_impl]
impl Client {
pub fn init() -> PhpResult<u64>{
pub fn init() -> PhpResult<u64> {
EventLoop::init()
}
pub fn wakeup() -> PhpResult<()> {
EventLoop::wakeup()
}
pub async fn get(url: &str) -> anyhow::Result<String> {
Ok(reqwest::get(url)
.await?
.text()
.await?
)
Ok(reqwest::get(url).await?.text().await?)
}
}
@ -29,7 +24,5 @@ pub extern "C" fn request_shutdown(_type: i32, _module_number: i32) -> i32 {
#[php_module]
pub fn get_module(module: ModuleBuilder) -> ModuleBuilder {
module
.request_shutdown_function(request_shutdown)
module.request_shutdown_function(request_shutdown)
}

11
src/borrow_unchecked.rs
View File

@ -5,21 +5,20 @@
//
// This is needed because of https://github.com/danog/php-tokio/blob/master/src/event_loop.rs#L72
//
// Rust thinks we're Sending the Future to another thread (tokio's event loop),
// Rust thinks we're Sending the Future to another thread (tokio's event loop),
// where it may be used even after its lifetime expires in the main (PHP) thread.
//
// In reality, the Future is only used by Tokio until the result is ready.
//
// Rust does not understand that when we suspend the current fiber in suspend_on,
// we basically keep alive the the entire stack,
// Rust does not understand that when we suspend the current fiber in suspend_on,
// we basically keep alive the the entire stack,
// including the Rust stack and the Future on it, until the result of the future is ready.
//
// Once the result of the Future is ready, tokio doesn't need it anymore,
// Once the result of the Future is ready, tokio doesn't need it anymore,
// the suspend_on function is resumed, and we safely drop the Future upon exiting.
use nicelocal_ext_php_rs::binary_slice::{BinarySlice, PackSlice};
#[inline(always)]
pub unsafe fn borrow_unchecked<
'original,
@ -69,4 +68,4 @@ unsafe impl<'original, 'unbounded, T: 'unbounded + PackSlice> BorrowUnchecked<'o
unsafe fn borrow_unchecked(self) -> Self::Unbounded {
unsafe { ::core::mem::transmute(self) }
}
}
}

101
src/event_loop.rs
View File

@ -1,20 +1,19 @@
use std::cell::RefCell;
use std::fs::File;
use std::future::Future;
use std::io::{self, Write};
use std::os::fd::{RawFd, FromRawFd};
use std::sync::mpsc::{Sender, Receiver, channel};
use std::io::Read;
use crate::borrow_unchecked::borrow_unchecked;
use lazy_static::lazy_static;
use nicelocal_ext_php_rs::boxed::ZBox;
use nicelocal_ext_php_rs::call_user_func;
use nicelocal_ext_php_rs::prelude::*;
use nicelocal_ext_php_rs::types::ZendHashTable;
use nicelocal_ext_php_rs::zend::Function;
use lazy_static::lazy_static;
use tokio::runtime::Runtime;
use std::cell::RefCell;
use std::fs::File;
use std::future::Future;
use std::io::Read;
use std::io::{self, Write};
use std::os::fd::AsRawFd;
use std::os::fd::{FromRawFd, RawFd};
use std::sync::mpsc::{channel, Receiver, Sender};
use tokio::runtime::Runtime;
lazy_static! {
pub static ref RUNTIME: Runtime = Runtime::new().expect("Could not allocate runtime");
@ -51,66 +50,75 @@ pub struct EventLoop {
impl EventLoop {
pub fn init() -> PhpResult<u64> {
EVENTLOOP.with_borrow_mut(|e| {
Ok(
match e {
None => e.insert(Self::new()?),
Some(ev) => ev
}.notify_receiver.as_raw_fd() as u64
)
Ok(match e {
None => e.insert(Self::new()?),
Some(ev) => ev,
}
.notify_receiver
.as_raw_fd() as u64)
})
}
pub fn suspend_on<T: Send + 'static, F: Future<Output = T> + Send + 'static>(future: F) -> T {
// What's going on here? Unsafe borrows???
// NO: this is actually 100% safe, and here's why.
//
// Rust thinks we're Sending the Future to another thread (tokio's event loop),
// Rust thinks we're Sending the Future to another thread (tokio's event loop),
// where it may be used even after its lifetime expires in the main (PHP) thread.
//
// In reality, the Future is only used by Tokio until the result is ready.
//
// Rust does not understand that when we suspend the current fiber in suspend_on,
// we basically keep alive the the entire stack,
// Rust does not understand that when we suspend the current fiber in suspend_on,
// we basically keep alive the the entire stack,
// including the Rust stack and the Future on it, until the result of the future is ready.
//
// Once the result of the Future is ready, tokio doesn't need it anymore,
// Once the result of the Future is ready, tokio doesn't need it anymore,
// the suspend_on function is resumed, and we safely drop the Future upon exiting.
//
let (future, get_current_suspension) = EVENTLOOP.with_borrow_mut(move |c| {
let c = c.as_mut().unwrap();
let idx = c.fibers.len() as u64;
c.fibers.insert_at_index(idx, call_user_func!(c.get_current_suspension).unwrap()).unwrap();
c.fibers
.insert_at_index(idx, call_user_func!(c.get_current_suspension).unwrap())
.unwrap();
let sender = c.sender.clone();
let mut notifier = c.notify_sender.try_clone().unwrap();
(RUNTIME.spawn(async move {
let res = future.await;
sender.send(idx).unwrap();
notifier.write_all(&[0]).unwrap();
res
}), unsafe {
borrow_unchecked(&c.get_current_suspension)
})
(
RUNTIME.spawn(async move {
let res = future.await;
sender.send(idx).unwrap();
notifier.write_all(&[0]).unwrap();
res
}),
unsafe { borrow_unchecked(&c.get_current_suspension) },
)
});
// We suspend the fiber here, the Rust stack is kept alive until the result is ready.
call_user_func!(get_current_suspension).unwrap().try_call_method("suspend", vec![]).unwrap();
call_user_func!(get_current_suspension)
.unwrap()
.try_call_method("suspend", vec![])
.unwrap();
// We've resumed, the `future` is already resolved and is not used by the tokio thread, it's safe to drop it.
return RUNTIME.block_on(future).unwrap();
}
pub fn wakeup() -> PhpResult<()> {
EVENTLOOP.with_borrow_mut(|c| {
let c = c.as_mut().unwrap();
c.notify_receiver.read_exact(&mut c.dummy).unwrap();
for fiber_id in c.receiver.try_iter() {
if let Some(fiber) = c.fibers.get_index_mut(fiber_id) {
fiber.object_mut().unwrap().try_call_method("resume", vec![])?;
fiber
.object_mut()
.unwrap()
.try_call_method("resume", vec![])?;
c.fibers.remove_index(fiber_id);
}
}
@ -127,10 +135,22 @@ impl EventLoop {
let (notify_receiver, notify_sender) =
sys_pipe().map_err(|err| format!("Could not create pipe: {}", err))?;
if !call_user_func!(Function::from_function("class_exists"), "\\Revolt\\EventLoop")?.bool().unwrap_or(false) {
if !call_user_func!(
Function::from_function("class_exists"),
"\\Revolt\\EventLoop"
)?
.bool()
.unwrap_or(false)
{
return Err(format!("\\Revolt\\EventLoop does not exist!").into());
}
if !call_user_func!(Function::from_function("interface_exists"), "\\Revolt\\EventLoop\\Suspension")?.bool().unwrap_or(false) {
if !call_user_func!(
Function::from_function("interface_exists"),
"\\Revolt\\EventLoop\\Suspension"
)?
.bool()
.unwrap_or(false)
{
return Err(format!("\\Revolt\\EventLoop\\Suspension does not exist!").into());
}
@ -141,8 +161,11 @@ impl EventLoop {
notify_sender: unsafe { File::from_raw_fd(notify_sender) },
notify_receiver: unsafe { File::from_raw_fd(notify_receiver) },
dummy: [0; 1],
get_current_suspension: Function::try_from_method("\\Revolt\\EventLoop", "getSuspension").ok_or("\\Revolt\\EventLoop::getSuspension does not exist")?,
get_current_suspension: Function::try_from_method(
"\\Revolt\\EventLoop",
"getSuspension",
)
.ok_or("\\Revolt\\EventLoop::getSuspension does not exist")?,
})
}
}

4
src/lib.rs
View File

@ -1,6 +1,6 @@
pub mod borrow_unchecked;
mod event_loop;
pub use php_tokio_derive::php_async_impl;
pub use event_loop::RUNTIME;
pub use event_loop::EventLoop;
pub use event_loop::RUNTIME;
pub use php_tokio_derive::php_async_impl;