Merge pull request #15 from sazima/task

fix: change PeriodicCallback -> time.sleep

Author: sazima

client/heart_beat_task.py

@@ -1,9 +1,8 @@
 import asyncio
-import threading
+import logging
 import time
 import traceback
 
-from tornado import ioloop
 
 from common import websocket
 from common.logger_factory import LoggerFactory
@@ -16,30 +15,38 @@
 
 
 class HeatBeatTask:
- def __init__(self, ws: websocket.WebSocketApp):
+ def __init__(self, ws: websocket.WebSocketApp, sleep_break: int,):
  self.ws: websocket.WebSocketApp = ws
  self.is_running = False
  self.recv_heart_beat_time: float = time.time()
- self.lock = asyncio.Lock()
+ self.sleep_break = sleep_break
 
  def set_recv_heart_beat_time(self, d: float):
  self.recv_heart_beat_time = d
 
- async def run(self):
- if self.lock.locked():
- LoggerFactory.get_logger().error('locked return')
- return
- async with self.lock:
+ def run(self):
+ # asyncio.set_event_loop(self.event_loop)
+ while True:
+ time.sleep(self.sleep_break)
+ if not self.is_running:
+ continue
+ if LoggerFactory.get_logger().isEnabledFor(logging.DEBUG):
+ LoggerFactory.get_logger().debug('run send heartbeat')
  try:
- await asyncio.wait_for(ioloop.IOLoop.current().run_in_executor(None, self.send_heart_beat), timeout=20)
+ # await asyncio.wait_for(asyncio.get_event_loop().run_in_executor(None, self.send_heart_beat), timeout=20)
+ self.send_heart_beat()
  except WebSocketConnectionClosedException:
  try:
- await asyncio.wait_for(ioloop.IOLoop.current().run_in_executor(None, self._close_and_on_close), timeout=20)
+ self._close_and_on_close()
+ # await asyncio.wait_for(asyncio.get_event_loop().run_in_executor(None, self._close_and_on_close), timeout=20)
  except Exception:
  LoggerFactory.get_logger().error(traceback.format_exc())
  except Exception:
  LoggerFactory.get_logger().error(traceback.format_exc())
- self.check_recv_heart_beat_time()
+ try:
+ self.check_recv_heart_beat_time()
+ except Exception:
+ LoggerFactory.get_logger().error(traceback.format_exc())
 
  def send_heart_beat(self):
  if self.is_running:

client/tcp_forward_client.py

@@ -51,6 +51,7 @@ def handle_message(self, each: socket.socket, data: ResisterAppendData):
  if data.speed_limiter:
  data.speed_limiter.add(len(recv))
  except OSError as e:
+ LoggerFactory.get_logger().error(f'{traceback.format_exc()}')
  LoggerFactory.get_logger().warn(f'get os error: {str(e)}, close')
  recv = b''
  send_message: MessageEntity = {

constant/system_constant.py

@@ -15,6 +15,6 @@ class SystemConstant:
 
  COOKIE_EXPIRE_SECONDS = 3600 * 24
 
- VERSION = '1.1.22'
+ VERSION = '1.1.23'
 
  GITHUB = 'https://github.com/sazima/proxynt'

run_client.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import logging
 import os
@@ -26,7 +27,6 @@
 from common.speed_limit import SpeedLimiter
 from common.websocket import WebSocketException, ABNF, WebSocketConnectionClosedException
 
-from client.clear_nonce_task import ClearNonceTask
 from client.heart_beat_task import HeatBeatTask
 from client.tcp_forward_client import TcpForwardClient
 from common import websocket
@@ -246,13 +246,14 @@ def main():
  url += '?c=' + json.dumps(compress_support)
  ws = websocket.WebSocketApp(url)
  forward_client = TcpForwardClient(ws, compress_support)
- heart_beat_task = HeatBeatTask(ws)
+ heart_beat_task = HeatBeatTask(ws, SystemConstant.HEART_BEAT_INTERVAL)
  WebsocketClient(ws, forward_client, heart_beat_task, config_data)
  LoggerFactory.get_logger().info('start run_forever')
  Thread(target=run_client, args=(ws,)).start() # 为了使用tornado的ioloop 方便设置超时
- ioloop.PeriodicCallback(heart_beat_task.run, SystemConstant.HEART_BEAT_INTERVAL * 1000).start()
- clear_nonce_stak = ClearNonceTask()
- ioloop.PeriodicCallback(clear_nonce_stak.run, 1800 * 1000).start()
+ Thread(target=heart_beat_task.run).start()
+ # ioloop.PeriodicCallback(heart_beat_task.run, SystemConstant.HEART_BEAT_INTERVAL * 1000).start()
+ # clear_nonce_stak = ClearNonceTask()
+ # ioloop.PeriodicCallback(clear_nonce_stak.run, 1800 * 1000).start()
  ioloop.IOLoop.current().start()
 
 

run_server.py

@@ -6,7 +6,7 @@
 import sys
 import time
 from optparse import OptionParser
-
+from threading import Thread
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
@@ -130,13 +130,13 @@ def main():
  app = tornado.web.Application(handlers, static_path=static_path, static_url_prefix=status_url_path, template_path=template_path)
  app.listen(ContextUtils.get_port(), chunk_size=65536 * 2)
  LoggerFactory.get_logger().info(f'start server at port {ContextUtils.get_port()}, websocket_path: {websocket_path}, admin_path: {admin_html_path}')
- heart_beat_task = HeartBeatTask(asyncio.get_event_loop())
  # create interval task
+ heart_beat_task = HeartBeatTask(asyncio.get_event_loop(), SystemConstant.HEART_BEAT_INTERVAL)
+ Thread(target=heart_beat_task.run).start()
  check_cookie_task = CheckCookieTask()
- tornado.ioloop.PeriodicCallback(heart_beat_task.run, SystemConstant.HEART_BEAT_INTERVAL * 1000).start()
  tornado.ioloop.PeriodicCallback(check_cookie_task.run, 3600 * 1000).start()
- clear_nonce_stak = ClearNonceTask()
- tornado.ioloop.PeriodicCallback(clear_nonce_stak.run, 1800 * 1000).start()
+ # clear_nonce_stak = ClearNonceTask()
+ # tornado.ioloop.PeriodicCallback(clear_nonce_stak.run, 1800 * 1000).start()
  tornado.ioloop.IOLoop.current().start()
 
 

server/task/heart_beat_task.py

@@ -12,15 +12,21 @@
 
 
 class HeartBeatTask:
- def __init__(self, loop):
+ def __init__(self, loop, break_time):
  self.loop = loop
+ self.break_time = break_time
 
- async def run(self):
- try:
- await asyncio.wait_for(asyncio.get_event_loop().run_in_executor(None, self.send_heart_beat), timeout=20)
- except Exception:
- LoggerFactory.get_logger().error(traceback.format_exc())
- self.check_recv_heart_beat_time()
+ def run(self):
+ while True:
+ time.sleep(self.break_time)
+ try:
+ self.send_heart_beat()
+ except Exception:
+ LoggerFactory.get_logger().error(traceback.format_exc())
+ try:
+ self.check_recv_heart_beat_time()
+ except Exception:
+ LoggerFactory.get_logger().error(traceback.format_exc())
 
  def send_heart_beat(self):
  asyncio.set_event_loop(self.loop)