mirrors/ntex
1
0
Fork 0
mirror of https://github.com/ntex-rs/ntex.git synced 2025-04-04 13:27:39 +03:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Nikolay Kim 2025-03-15 01:08:34 +05:00
parent 5d29038f30
commit f240db8296
1 changed files with 55 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

95
ntex-net/src/rt_polling/driver.rs
View file

@ -121,48 +121,42 @@ impl<T> Handler for StreamOpsHandler<T> {
for (id, change) in self.feed.drain(..) { for (id, change) in self.feed.drain(..) {
match change { match change {
Change::Readable => { Change::Readable => {
if let Some(item) = streams.get_mut(id) { let item = &mut streams[id];
let result = item.context.with_read_buf(|buf| { let result = item.context.with_read_buf(|buf| {
let chunk = buf.chunk_mut(); let chunk = buf.chunk_mut();
let b = chunk.as_mut_ptr(); let b = chunk.as_mut_ptr();
Poll::Ready( Poll::Ready(
task::ready!(syscall!( task::ready!(syscall!(
break libc::read(item.fd, b as _, chunk.len()) break libc::read(item.fd, b as _, chunk.len())
)) ))
.inspect(|size| { .inspect(|size| {
unsafe { buf.advance_mut(*size) }; unsafe { buf.advance_mut(*size) };
log::debug!( log::debug!(
"{}: {:?}, SIZE: {:?}, BUF: {:?}", "{}: {:?}, SIZE: {:?}, BUF: {:?}",
item.context.tag(), item.context.tag(),
item.fd, item.fd,
size, size,
buf, buf,
); );
}), }),
) )
}); });
if result.is_pending() { if result.is_pending() {
self.inner.api.register(item.fd, id, Interest::Readable); self.inner.api.register(item.fd, id, Interest::Readable);
}
} }
} }
Change::Writable => { Change::Writable => {
if let Some(item) = streams.get_mut(id) { let item = &mut streams[id];
let result = item.context.with_write_buf(|buf| { let result = item.context.with_write_buf(|buf| {
let slice = &buf[..]; let slice = &buf[..];
syscall!( syscall!(
break libc::write( break libc::write(item.fd, slice.as_ptr() as _, slice.len())
item.fd, )
slice.as_ptr() as _, });
slice.len()
)
)
});
if result.is_pending() { if result.is_pending() {
self.inner.api.register(item.fd, id, Interest::Writable); self.inner.api.register(item.fd, id, Interest::Writable);
}
} }
} }
Change::Error(err) => { Change::Error(err) => {
@ -259,9 +253,30 @@ impl<T> StreamCtl<T> {
self.id, self.id,
item.fd item.fd
); );
self.inner
.api let result = item.context.with_read_buf(|buf| {
.register(item.fd, self.id, Interest::Readable); let chunk = buf.chunk_mut();
let b = chunk.as_mut_ptr();
Poll::Ready(
task::ready!(syscall!(break libc::read(item.fd, b as _, chunk.len())))
.inspect(|size| {
unsafe { buf.advance_mut(*size) };
log::debug!(
"{}: {:?}, SIZE: {:?}, BUF: {:?}",
item.context.tag(),
item.fd,
size,
buf,
);
}),
)
});
if result.is_pending() {
self.inner
.api
.register(item.fd, self.id, Interest::Readable);
}
}) })
} }