From 24ea2c6bdcdf54fd35725865827bcbb908a44b02 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sun, 17 Dec 2023 12:21:17 +0800 Subject: [PATCH] bugfix --- appPublic/sshx.py | 64 +++++++++++++++++++++++++-------------------- appPublic/worker.py | 28 ++++++++++++-------- 2 files changed, 53 insertions(+), 39 deletions(-) diff --git a/appPublic/sshx.py b/appPublic/sshx.py index e8a9b46..2eb3a7e 100644 --- a/appPublic/sshx.py +++ b/appPublic/sshx.py @@ -1,5 +1,6 @@ import os import sys +import time import shlex from functools import partial from threading import Thread @@ -18,6 +19,7 @@ class SSHNode: "password":password, "port":port } + print(self.server2) self.jumpers = jumpers self.conn = None self.jumper_conns = [] @@ -107,48 +109,54 @@ class SSHNode: term_size=(80,24), encoding='utf-8' ) - buffer = '' - msglen = len(xmsgs) - for i in range(msglen): - if xmsgs[i][0]: - break - s = xmsgs[i][1].format(**ns) - proc.stdin.write(s) keyin = False def feed_data(xmsgs, debug_input): if len(xmsgs) == 0: + print('#####++##### xmsgs has zero elements') return - a = xmsgs[0] - xmsgs.reverse() - xmsgs.pop() - xmsgs.reverse() - s = a[1].format(**ns) - proc.stdin.write(s) - if debug_input: - debug_input(f'{s=},{a=}') keyin = True + a = xmsgs.pop(0) + while True: + if a[1] is None: + proc.stdin.write_eof() + self.running = False + else: + s = a[1].format(**ns) + proc.stdin.write(s) + + if len(xmsgs) == 0 or xmsgs[0][0]: + break + a = xmsgs.pop(0) already_output = False callee = None loop = asyncio.get_event_loop() - while True: - if proc.stdout.at_eof(): - break - + self.running = True + while self.running: if keyin: keyin = False await proc.stdin.drain() - - if len(xmsgs) > 0 and already_output: - f = partial(feed_data, xmsgs, show_input) - callee = loop.call_later(xmsgs[0][0], f) - x = await proc.stdout.read(1024) - if callee: - callee.cancel() - if show_stdout: + if proc.stdout.at_eof(): + break + tup = proc.collect_output() + x = tup[0] + if x!='' and show_stdout: + if x is None: + break + if callee: + callee.cancel() + callee = None show_stdout(x) - already_output = True + else: + if callee is None: + if len(xmsgs) > 0: + f = partial(feed_data, xmsgs, show_input) + t = xmsgs[0][0] or 0 + callee = loop.call_later(t, f) + await asyncio.sleep(0.05) + + print('##########fininshed##########') async def _run(self, cmd, input=None, stdin=None, stdout=None): if cmd.startswith('l2r'): diff --git a/appPublic/worker.py b/appPublic/worker.py index e25cce4..152b3fc 100755 --- a/appPublic/worker.py +++ b/appPublic/worker.py @@ -1,21 +1,27 @@ import random import asyncio +import inspect from functools import wraps -def asyncCall(func): +def to_func(func): @wraps(func) def wraped_func(*args,**kw): - task = asyncio.ensure_future(func(*args,**kw)) - asyncio.gather(task) + if inspect.iscoroutinefunction(func): + task = asyncio.ensure_future(func(*args,**kw)) + ret = asyncio.gather(task) + return ret + return func(*args, **kw) return wraped_func -class Worker: +class AsyncWorker: def __init__(self,maxtask=50): self.semaphore = asyncio.Semaphore(maxtask) async def __call__(self,callee,*args,**kw): async with self.semaphore: - return await callee(*args,**kw) + if inspect.iscoroutinefunction(callee): + return await callee(*args,**kw) + return callee(*args, **kw) async def run(self,cmd): async with self.semaphore: @@ -27,16 +33,16 @@ class Worker: return stdout, stderr if __name__ == '__main__': - async def hello(cnt,greeting): + def hello(cnt,greeting): t = random.randint(1,10) - await asyncio.sleep(t) + # await asyncio.sleep(t) print(cnt,'cost ',t,'seconds to',greeting) - async def run(): - w = Worker() - tasks = [ w(hello,i,'hello world') for i in range(1000) ] - await asyncio.wait(tasks) + w = AsyncWorker() + g = [ asyncio.create_task(w(hello,i,'hello world')) for i in range(1000) ] + await asyncio.wait(g) + print('aaaaaaaaaaaaaaaaaaa') loop = asyncio.get_event_loop() loop.run_until_complete(run())