fixed corking in CPython + change uv_timer to uv_check to run asyncio

pull/39/head
Ciro 2022-10-25 12:33:50 -03:00
rodzic b825d81fbf
commit 8c4ffc918a
6 zmienionych plików z 79 dodań i 9 usunięć

Wyświetl plik

@ -2,6 +2,7 @@
import asyncio import asyncio
import threading import threading
import time import time
from queue import Queue
from .native import UVLoop from .native import UVLoop
@ -26,6 +27,8 @@ class Loop:
def __init__(self, exception_handler=None): def __init__(self, exception_handler=None):
self.loop = asyncio.new_event_loop() self.loop = asyncio.new_event_loop()
self.uv_loop = UVLoop() self.uv_loop = UVLoop()
self.queue = Queue()
if hasattr(exception_handler, '__call__'): if hasattr(exception_handler, '__call__'):
self.exception_handler = exception_handler self.exception_handler = exception_handler
self.loop.set_exception_handler(lambda loop, context: exception_handler(loop, context, None)) self.loop.set_exception_handler(lambda loop, context: exception_handler(loop, context, None))
@ -39,13 +42,25 @@ class Loop:
def set_timeout(self, timeout, callback, user_data): def set_timeout(self, timeout, callback, user_data):
return self.uv_loop.create_timer(timeout, 0, callback, user_data) return self.uv_loop.create_timer(timeout, 0, callback, user_data)
def enqueue(self, callback, user_data):
self.queue.put((callback, user_data))
def create_future(self): def create_future(self):
return self.loop.create_future() return self.loop.create_future()
def start(self): def start(self):
self.started = True self.started = True
#start relaxed until first task #run asyncio once per tick
self.timer = self.uv_loop.create_timer(0, 1, lambda loop: loop.run_once_asyncio(), self) def tick(loop):
#only call one item of the queue per tick
if not loop.queue.empty():
(callback, user_data) = loop.queue.get(False)
callback(user_data)
loop.queue.task_done()
#run once asyncio
loop.run_once_asyncio()
#use check for calling asyncio once per tick
self.timer = self.uv_loop.create_check(tick, self)
def run(self): def run(self):
self.uv_loop.run() self.uv_loop.run()

Wyświetl plik

@ -15,6 +15,11 @@ void socketify_generic_timer_callback(uv_timer_t *timer){
loop_data->handler(loop_data->user_data); loop_data->handler(loop_data->user_data);
} }
void socketify_generic_check_callback(uv_check_t *timer){
socketify_timer* loop_data = (socketify_timer*)uv_handle_get_data((uv_handle_t*)timer);
loop_data->handler(loop_data->user_data);
}
void* socketify_get_native_loop(socketify_loop* loop){ void* socketify_get_native_loop(socketify_loop* loop){
return loop->uv_loop; return loop->uv_loop;
@ -104,7 +109,7 @@ socketify_timer* socketify_create_timer(socketify_loop* loop, uint64_t timeout,
timer->handler = handler; timer->handler = handler;
uv_handle_set_data((uv_handle_t*)uv_timer, timer); uv_handle_set_data((uv_handle_t*)uv_timer, timer);
uv_timer_start(uv_timer, socketify_generic_timer_callback, timeout, repeat); uv_timer_start(uv_timer, socketify_generic_timer_callback, timeout, repeat);
return timer; return timer;
} }
@ -114,10 +119,37 @@ void socketify_timer_set_repeat(socketify_timer* timer, uint64_t repeat){
} }
//stops and destroy timer info //stops and destroy timer info
void socketify_timer_destroy(socketify_timer* timer){ void socketify_timer_destroy(socketify_timer* timer){
uv_timer_stop(timer->uv_timer_ptr); uv_timer_stop(timer->uv_timer_ptr);
free(timer->uv_timer_ptr); free(timer->uv_timer_ptr);
free(timer); free(timer);
}
socketify_timer* socketify_create_check(socketify_loop* loop, socketify_timer_handler handler, void* user_data){
uv_check_t* uv_timer = malloc(sizeof(uv_check_t));
if(uv_check_init(loop->uv_loop, uv_timer)){
free(uv_timer);
return NULL;
}
socketify_timer* timer = malloc(sizeof(socketify_timer));
timer->uv_timer_ptr = uv_timer;
timer->user_data = user_data;
timer->handler = handler;
uv_handle_set_data((uv_handle_t*)uv_timer, timer);
uv_check_start(uv_timer, socketify_generic_check_callback);
return timer;
}
//stops and destroy timer info
void socketify_check_destroy(socketify_timer* timer){
uv_check_stop(timer->uv_timer_ptr);
free(timer->uv_timer_ptr);
free(timer);
} }

Wyświetl plik

@ -42,4 +42,6 @@ socketify_timer* socketify_create_timer(socketify_loop* loop, uint64_t timeout,
void socketify_timer_destroy(socketify_timer* timer); void socketify_timer_destroy(socketify_timer* timer);
void socketify_timer_set_repeat(socketify_timer* timer, uint64_t repeat); void socketify_timer_set_repeat(socketify_timer* timer, uint64_t repeat);
socketify_timer* socketify_create_check(socketify_loop* loop, socketify_timer_handler handler, void* user_data);
void socketify_check_destroy(socketify_timer* timer);
#endif #endif

Wyświetl plik

@ -50,6 +50,9 @@ void socketify_timer_destroy(socketify_timer* timer);
bool socketify_async_call(socketify_loop* loop, socketify_async_handler handler, void* user_data); bool socketify_async_call(socketify_loop* loop, socketify_async_handler handler, void* user_data);
void socketify_timer_set_repeat(socketify_timer* timer, uint64_t repeat); void socketify_timer_set_repeat(socketify_timer* timer, uint64_t repeat);
socketify_timer* socketify_create_check(socketify_loop* loop, socketify_timer_handler handler, void* user_data);
void socketify_check_destroy(socketify_timer* timer);
""") """)
library_path = os.path.join(os.path.dirname(__file__), "libsocketify.so") library_path = os.path.join(os.path.dirname(__file__), "libsocketify.so")
@ -60,12 +63,26 @@ def socketify_generic_handler(data):
if not data == ffi.NULL: if not data == ffi.NULL:
(handler, user_data) = ffi.from_handle(data) (handler, user_data) = ffi.from_handle(data)
handler(user_data) handler(user_data)
class UVCheck:
def __init__(self, loop, handler, user_data):
self._handler_data = ffi.new_handle((handler, user_data))
self._ptr = lib.socketify_create_check(loop, socketify_generic_handler, self._handler_data)
def stop(self):
lib.socketify_check_destroy(self._ptr)
self._handler_data = None
self._ptr = ffi.NULL
def __del__(self):
if self._ptr != ffi.NULL:
lib.socketify_check_destroy(self._ptr)
self._handler_data = None
class UVTimer: class UVTimer:
def __init__(self, loop, timeout, repeat, handler, user_data): def __init__(self, loop, timeout, repeat, handler, user_data):
self._handler_data = ffi.new_handle((handler, user_data)) self._handler_data = ffi.new_handle((handler, user_data))
self._ptr = lib.socketify_create_timer(loop, ffi.cast("uint64_t", timeout), ffi.cast("uint64_t", repeat), socketify_generic_handler, self._handler_data) self._ptr = lib.socketify_create_timer(loop, ffi.cast("uint64_t", timeout), ffi.cast("uint64_t", repeat), socketify_generic_handler, self._handler_data)
def stop(self): def stop(self):
lib.socketify_timer_destroy(self._ptr) lib.socketify_timer_destroy(self._ptr)
self._handler_data = None self._handler_data = None
@ -93,6 +110,9 @@ class UVLoop:
def create_timer(self, timeout, repeat, handler, user_data): def create_timer(self, timeout, repeat, handler, user_data):
return UVTimer(self._loop, timeout, repeat, handler, user_data) return UVTimer(self._loop, timeout, repeat, handler, user_data)
def create_check(self, handler, user_data):
return UVCheck(self._loop, handler, user_data)
def prepare_unbind(self): def prepare_unbind(self):
lib.socketify_prepare_unbind(self._loop) lib.socketify_prepare_unbind(self._loop)

Wyświetl plik

@ -418,11 +418,12 @@ class AppResponse:
def cork(self, callback): def cork(self, callback):
if not self.aborted: if not self.aborted:
self._cork_handler = callback self._cork_handler = callback
if is_python: if is_python: #call is enqueued to garantee corking works properly in python3
lib.uws_res_cork(self.SSL, self.res, uws_generic_cork_handler, self._ptr) self.loop.enqueue(lambda instance: lib.uws_res_cork(instance.SSL, instance.res, uws_generic_cork_handler, instance._ptr), self)
else: #just add to uv loop in next tick to garantee corking works properly in pypy3 else: #just add to uvloop in next tick to garantee corking works properly in pypy3
self.loop.set_timeout(0, lambda instance: lib.uws_res_cork(instance.SSL, instance.res, uws_generic_cork_handler, instance._ptr), self) self.loop.set_timeout(0, lambda instance: lib.uws_res_cork(instance.SSL, instance.res, uws_generic_cork_handler, instance._ptr), self)
def set_cookie(self, name, value, options={}): def set_cookie(self, name, value, options={}):
if self._write_jar == None: if self._write_jar == None:
self._write_jar = cookies.SimpleCookie() self._write_jar = cookies.SimpleCookie()

Wyświetl plik

@ -28,7 +28,7 @@ import asyncio
async def home(res, req): async def home(res, req):
# res.write_header("Content-Type", "plain/text") # res.write_header("Content-Type", "plain/text")
# await asyncio.sleep(0) await asyncio.sleep(0)
def corked(res): def corked(res):
res.write("Test ") res.write("Test ")