a=[1, 2.1]
b=["a", ["b", "c"]]
c=["a", {"b": 1}]
obj = rtoml.load("""
a=[1, 2.1]
""")
print(obj)
python3.6 celery5.0.5 windows 32位
# 1.send_task返回AsyncResult
# 2.AsyncResult的get()等待返回结果
# 3.get()会比真实多耗费1秒，并且每次请求都会出现
from celery import Celery
class xxxCelery(Celery):
def call_xxx(self, name, timeout=120, **kwargs):
LOG.info("send_task: %s" % locals())
start = time.time()
r = self.send_task(name, **kwargs)
g = eventlet.spawn(r.get, timeout=timeout)
result = g.wait()
print("cost: %s" % time.time()-start)
return
#celery/backends/asynchronuse.py
class greenletDrainer(Drainer):
...
def run(self):
self._started.set()
while not self._stopped.is_set():
try:
self.result_consumer.drain_events(timeout=1)
# 新增了两句, 估摸着是这个问题
self._send_drain_complete_event()
self._create_drain_complete_event()
except socket.timeout:
pass
self._shutdown.set()
import serial
from eventlet import tpool
class SerialSocket(serial.Serial):
def recv(self):
# 读完
if self.in_waiting:
return tpool.execute(self.read, self.in_waiting)
return data
def listen_message():
ss = SerialSocket("com10")
buffer = b''
while True:
data = ss.recv()
if data is not None:
buffer +=data
import serial
from eventlet import tpool, sleep
class SerialSocket(serial.Serial):
def recv(self):
# 读完
if self.in_waiting:
return tpool.execute(self.read, self.in_waiting)
return data
def listen_message():
ss = SerialSocket("com10")
buffer = b''
while True:
data = ss.recv()
if data is not None:
buffer +=data
else:
sleep(0.01)
import serial
from eventlet import tpool
class SerialSocket(serial.Serial):
def recv(self):
# 等待
data = tpool.execute(self.read, 1)
# 读完
while self.in_waiting:
data += tpool.execute(self.read, self.in_waiting)
return data
def listen_message():
ss = SerialSocket("com10")
buffer = b''
while True:
g = evenlet.spawn(ss.recv)
data = g.wait()
buffer +=data
def start_new_thread(function, args=(), kwargs=None):
...
print("hello: ", locals())
g = greenthread.spawn_n(__thread_body, function, args, kwargs)
return get_ident(g)
# -*- coding:utf-8 -*-
# from eventlet import patcher
# original_threading = patcher.original('threading')
import threading as original_threading
import time
import eventlet
eventlet.monkey_patch()
def thread1_run():
while True:
print("thread1:", time.time())
time.sleep(1)
if __name__ == "__main__":
thread1 = original_threading.Thread(target=thread1_run, name="thread1", daemon=True)
print(type(thread1))
thread1.start()
e = eventlet.Event()
e.wait()
class BaseRotatingHandler(logging.FileHandler):
def emit(self, record):
"""
Emit a record.
Output the record to the file, catering for rollover as described
in doRollover().
"""
try:
if self.shouldRollover(record):
self.doRollover()
logging.FileHandler.emit(self, record)
except Exception:
self.handleError(record)
import os
import time
import datetime
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
class AutumnRotatingFileHandler(RotatingFileHandler):
"""
日志+大小+支持windows
"""
def __init__(self, filename, mode='a', maxBytes=0, backupCount=99, encoding=None, delay=False, backupDayCount=30):
filename = str(filename)
self.date_format = "%Y%m%d"
self.create_date = self._now_date()
self.backupCount = backupCount
# 保存原始文件名
self.filename = str(Path(filename).absolute())
self.backupDayCount = backupDayCount
RotatingFileHandler.__init__(self, self.last_file_name(filename), mode=mode, maxBytes=maxBytes, backupCount=backupCount, encoding=encoding, delay=delay)
def _now_date(self):
return time.strftime(self.date_format, time.localtime())
def doRollover(self):
"""
Do a rollover, as described in __init__().
"""
# 关闭当前文件
if self.stream:
self.stream.close()
self.stream = None
# 生成最新文件名
suffix = "."+ self._now_date()
self.baseFilename = str(self.filename) + suffix
if self.backupCount > 0:
for i in range(0, self.backupCount):
number_suffix = "{:0>2d}".format(i)
sfn = ".".join([self.baseFilename, number_suffix])
if os.path.exists(sfn):
continue
else:
break
number_suffix = "{:0>2d}".format(i)
self.baseFilename = ".".join([self.baseFilename, number_suffix])
# 删除过期文件
self.deleteExpiredFiles()
# 打开
self.stream = self._open()
self.create_date = self._now_date()
def shouldRollover(self, record):
"""
Determine if rollover should occur.
Basically, see if the supplied record would cause the file to exceed
the size limit we have.
"""
# 文件分割条件1: 日期变化
if self._now_date() != self.create_date:
return 1
if self.stream is None: # delay was set...
self.stream = self._open()
# 文件分割条件1: 文件大小超过限制
if self.maxBytes > 0: # are we rolling over?
msg = "%s\n" % self.format(record)
self.stream.seek(0, 2) #due to non-posix-compliant Windows feature
if self.stream.tell() + len(msg) >= self.maxBytes:
return 1
return 0
def deleteExpiredFiles(self):
"""
删除过期文件
"""
dead_datetime = datetime.datetime.now() - datetime.timedelta(days=self.backupDayCount)
filenames = []
for i in range(1, 3):
date = dead_datetime - datetime.timedelta(days=1)
date_str = date.strftime(self.date_format)
base_name = "%s.%s" % (self.filename, date_str)
filenames.append(base_name)
for j in range(1, self.backupCount):
sfn = "%s.%d" % (base_name, j)
filenames.append(sfn)
# delete file
for filename in filenames:
if os.path.exists(filename):
try:
os.remove(sfn)
except:
pass
def last_file_name(self, filename):
"""
文件名
"""
suffix = "."+self._now_date()
base_filename = str(filename) + suffix
if self.backupCount > 0:
for i in reversed(range(0, self.backupCount)):
number_suffix = "{:0>2d}".format(i)
sfn = ".".join([base_filename, number_suffix])
# 最近存在的文件
if os.path.exists(sfn):
break
else:
continue
number_suffix = "{:0>2d}".format(i)
base_filename = ".".join([base_filename, number_suffix])
return base_filename
exception: [WinError 10054] 远程主机强迫关闭了一个现有的连接。 ... ValueError: file descriptor cannot be a negative integer (-1)
def wait(self, seconds=None):
....
try:
r, w, er = select.select(reader_fds, writer_fds, all_fds, seconds)
# 捕捉异常并移除句柄-1
except ValueError as e:
self.remove_descriptor(-1)
return
except select.error as e:
if support.get_errno(e) == errno.EINTR:
return
elif support.get_errno(e) in BAD_SOCK:
self._remove_bad_fds()
return
else:
raise
import sys
import eventlet
# eventlet.monkey_patch()
httplib2 = eventlet.import_patched('httplib2')
print(httplib2)
print(httplib2.socket)
print("================")
for k,v in sys.modules.items():
if "socket" in k:
print(k, v)
# 打印内容
<module 'httplib2' from 'D:\\workspace\\venv\\xxx\\lib\\site-packages\\httplib2\\__init__.py'>
<module 'eventlet.green.socket' from 'D:\\workspace\\venv\\xxx\\lib\\site-packages\\eventlet\\green\\socket.py'>
================
_socket <module '_socket' from 'c:\\python\\python36\\DLLs\\_socket.pyd'>
socket <module 'socket' from 'c:\\python\\python36\\lib\\socket.py'>
__original_module_socket <module 'socket' from 'c:\\python\\python36\\lib\\socket.py'>
# 说明了什么？
绿化只是替换httplib2模块中的引用属性socket。并未改变sys.modules中的属性
from eventlet.green import socket
import eventlet
httplib2 = eventlet.import_patched('httplib2')
print(httplib2)
print(httplib2.socket)
# 打印
<module 'httplib2' from 'D:\\workspace\\venv\\autumn-secs\\lib\\site-packages\\httplib2\\__init__.py'>
<module 'eventlet.green.socket' from 'D:\\workspace\\venv\\autumn-secs\\lib\\site-packages\\eventlet\\green\\socket.py'>
pip install transitions
# 连接协议状态机
from transitions.extensions import HierarchicalMachine as Machine
from transitions.extensions.nesting import NestedState
class ConnectionStateMachine:
"""Connection state machine."""
def __init__(self, callbacks=None):
"""
:param callbacks: callbacks for the state machine
"""
self.callbacks = {}
# 定义状态
self.states = ["STATE_NOT_CONNECTED",
{
'name': "STATE_CONNECTED", # 状态名
'on_enter': self._on_enter_CONNECTED, # 进入状态触发
'on_exit': self._on_exit_CONNECTED, # 退出状态触发
'children': [ # 状态嵌套
"STATE_NOT_SELECTED",
{
'name': "STATE_SELECTED",
'on_enter': self._on_enter_CONNECTED_SELECTED
}
]
}]
# transition 1
self.machine = Machine(model=self, states=self.states, initial="STATE_NOT_CONNECTED", auto_transitions=False)
if callbacks:
self.callbacks = callbacks
# 定义状态变换
self.machine.add_transition('connect', "STATE_NOT_CONNECTED", "STATE_CONNECTED_NOT_SELECTED") # transition 2
self.machine.add_transition('disconnect', "STATE_CONNECTED", "STATE_NOT_CONNECTED") # transition 3
self.machine.add_transition('select', "STATE_CONNECTED_NOT_SELECTED", "STATE_CONNECTED_SELECTED") # transition 4
self.machine.add_transition('deselect', "STATE_CONNECTED_SELECTED", "STATE_CONNECTED_NOT_SELECTED") # transition 5
self.machine.add_transition('timeoutT7', "STATE_CONNECTED_NOT_SELECTED", "STATE_NOT_CONNECTED") # transition 6
# 事件触发的动作
def _on_enter_CONNECTED(self):
if "on_enter_CONNECTED" in self.callbacks:
self.callbacks["on_enter_CONNECTED"]()
def _on_exit_CONNECTED(self):
if "on_exit_CONNECTED" in self.callbacks:
self.callbacks["on_exit_CONNECTED"]()
def _on_enter_CONNECTED_SELECTED(self):
if "on_enter_CONNECTED_SELECTED" in self.callbacks:
self.callbacks["on_enter_CONNECTED_SELECTED"]()
# 一般都是两种：不嵌套、嵌套 from transitions import Machine #不嵌套 from transitions.extensions import HierarchicalMachine as Machine # 嵌套 # 检查状态 machine.state machine.is_<state_name>()
# 状态可以是三种类型: 对象、字符串、字典
from transitions import State
states = [
State(name="solid"), # 对象
"liquid", # 字符串
{"name": "gas"} # 字典
]
# 定义函数
machine.add_transition(trigger, source, dest, )
trigger(str) 方法名,触发状态转换
source(str or list) 源状态
dest(str) 目标状态
# 加上变换
machine.add_transition(trigger="wake_up", source="asleep", dest="hanging out")
machine.add_transition('work_out', 'hanging out', 'hungry')
machine.add_transition('nap', '*', 'asleep') # 从任意状态变为asleep
def main():
import eventlet
httplib2 = eventlet.import_patched('httplib2')
shared_resource = httplib2.Http()
def get_url():
resp, content = shared_resource.request("http://eventlet.net")
return content
p = eventlet.GreenPile()
p.spawn(get_url)
p.spawn(get_url)
results = list(p)
assert results[0] == results[1]
if __name__ == "__main__":
main()
Traceback (most recent call last):
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/hubs/poll.py", line 111, in wait
listener.cb(fileno)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/greenthread.py", line 221, in main
result = function(*args, **kwargs)
File "test.py", line 12, in get_url
resp, content = shared_resource.request("http://eventlet.net")
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/httplib2/__init__.py", line 1994, in request
cachekey,
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/httplib2/__init__.py", line 1651, in _request
conn, request_uri, method, body, headers
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/httplib2/__init__.py", line 1558, in _conn_request
conn.request(method, request_uri, body, headers)
File "/root/miniconda3/envs/defect/lib/python3.7/http/client.py", line 1252, in request
self._send_request(method, url, body, headers, encode_chunked)
File "/root/miniconda3/envs/defect/lib/python3.7/http/client.py", line 1298, in _send_request
self.endheaders(body, encode_chunked=encode_chunked)
File "/root/miniconda3/envs/defect/lib/python3.7/http/client.py", line 1247, in endheaders
self._send_output(message_body, encode_chunked=encode_chunked)
File "/root/miniconda3/envs/defect/lib/python3.7/http/client.py", line 1026, in _send_output
self.send(msg)
File "/root/miniconda3/envs/defect/lib/python3.7/http/client.py", line 987, in send
self.sock.sendall(data)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/greenio/base.py", line 403, in sendall
tail = self.send(data, flags)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/greenio/base.py", line 397, in send
return self._send_loop(self.fd.send, data, flags)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/greenio/base.py", line 392, in _send_loop
timeout_exc=_timeout_exc)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/greenio/base.py", line 210, in _trampoline
mark_as_closed=self._mark_as_closed)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/hubs/__init__.py", line 157, in trampoline
listener = hub.add(hub.WRITE, fileno, current.switch, current.throw, mark_as_closed)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/hubs/epolls.py", line 22, in add
listener = hub.BaseHub.add(self, evtype, fileno, cb, tb, mac)
File "/root/miniconda3/envs/defect/lib/python3.7/site-packages/eventlet/hubs/hub.py", line 181, in add
evtype, fileno, evtype, cb, bucket[fileno]))
RuntimeError: Second simultaneous write on fileno 5 detected. Unless you really know what you're doing, make sure that only one greenthread can write any particular socket. Consider using a pools.Pool. If you do know what you're doing and want to disable this error, call eventlet.debug.hub_prevent_multiple_readers(False) - MY THREAD=<built-in method switch of GreenThread object at 0x7f25df161680>; THAT THREAD=FdListener('write', 5, <built-in method switch of GreenThread object at 0x7f25df161470>, <built-in method throw of GreenThread object at 0x7f25df161470>)
def main():
import eventlet
# httplib2 = eventlet.import_patched('httplib2')
import http2
eventlet.monkey_patch()
shared_resource = httplib2.Http()
def get_url():
resp, content = shared_resource.request("http://eventlet.net")
return content
p = eventlet.GreenPile()
p.spawn(get_url)
p.spawn(get_url)
results = list(p)
assert results[0] == results[1]
if __name__ == "__main__":
main()
