RanchiMall/pyflo
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

connector

Browse Source
This commit is contained in:
4tochka 2019-05-16 08:59:15 +04:00
parent b60fe22663
commit c351166620
1 changed files with 21 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
pybtc/connector/block_loader.py
View File

@ -19,8 +19,6 @@ class BlockLoader:
def __init__(self, parent, workers=8): def __init__(self, parent, workers=8):
self.worker_limit = workers self.worker_limit = workers
self.worker = dict() self.worker = dict()
self.worker_reader = dict()
self.worker_writer = dict()
self.worker_tasks = list() self.worker_tasks = list()
self.worker_busy = dict() self.worker_busy = dict()
self.parent = parent self.parent = parent
@ -81,13 +79,9 @@ class BlockLoader:
self.worker_busy[i] = True self.worker_busy[i] = True
if height <= self.parent.last_block_height: if height <= self.parent.last_block_height:
height = self.parent.last_block_height + 1 height = self.parent.last_block_height + 1
# self.pipe_sent_msg(self.worker[i].writer, b'rpc_batch_limit', self.pipe_sent_msg(self.worker[i].writer, b'rpc_batch_limit',
# int_to_bytes(self.rpc_batch_limit))
# self.pipe_sent_msg(self.worker[i].writer, b'get', int_to_bytes(height))
self.pipe_sent_msg(self.worker_writer[i], b'rpc_batch_limit',
int_to_bytes(self.rpc_batch_limit)) int_to_bytes(self.rpc_batch_limit))
self.pipe_sent_msg(self.worker_writer[i], b'get', int_to_bytes(height)) self.pipe_sent_msg(self.worker[i].writer, b'get', int_to_bytes(height))
height += self.rpc_batch_limit height += self.rpc_batch_limit
new_requests += 1 new_requests += 1
if not new_requests: if not new_requests:
@ -113,28 +107,23 @@ class BlockLoader:
in_writer, out_writer = os.fdopen(in_writer,'wb'), os.fdopen(out_writer,'wb') in_writer, out_writer = os.fdopen(in_writer,'wb'), os.fdopen(out_writer,'wb')
# create new process # create new process
# worker = Process(target=Worker, args=(index, in_reader, in_writer, out_reader, out_writer, worker = Process(target=Worker, args=(index, in_reader, in_writer, out_reader, out_writer,
# self.rpc_url, self.rpc_timeout, self.rpc_batch_limit)) self.rpc_url, self.rpc_timeout, self.rpc_batch_limit))
# worker.start() worker.start()
k = self.loop.create_task(Worker(index, in_reader, in_writer, out_reader, out_writer,
self.rpc_url, self.rpc_timeout, self.rpc_batch_limit))
in_reader.close() in_reader.close()
out_writer.close() out_writer.close()
# get stream reader # get stream reader
# worker.reader = await self.get_pipe_reader(out_reader) worker.reader = await self.get_pipe_reader(out_reader)
# worker.writer = in_writer worker.writer = in_writer
# worker.name = str(index) worker.name = str(index)
self.worker_reader[index] = out_reader self.worker[index] = worker
self.worker_writer[index] = in_writer
# self.worker[index] = worker
self.worker_busy[index] = False self.worker_busy[index] = False
# start message loop # start message loop
self.loop.create_task(self.message_loop(index)) self.loop.create_task(self.message_loop(index))
# wait if process crash # wait if process crash
# r = await self.loop.run_in_executor(None, worker.join) await self.loop.run_in_executor(None, worker.join)
# del self.worker[index] del self.worker[index]
r =0 self.log.warning('Block loader worker %s is stopped' % index)
self.log.warning('Block loader worker %s is stopped [%s]' % (index, r))
@ -175,8 +164,7 @@ class BlockLoader:
async def message_loop(self, index): async def message_loop(self, index):
while True: while True:
# msg_type, msg = await self.pipe_get_msg(self.worker[index].reader) msg_type, msg = await self.pipe_get_msg(self.worker[index].reader)
msg_type, msg = await self.pipe_get_msg(self.worker_reader[index])
if msg_type == b'pipe_read_error': if msg_type == b'pipe_read_error':
if not self.worker[index].is_alive(): if not self.worker[index].is_alive():
return return
@ -206,7 +194,7 @@ class Worker:
def __init__(self, name , in_reader, in_writer, out_reader, out_writer, def __init__(self, name , in_reader, in_writer, out_reader, out_writer,
rpc_url, rpc_timeout, rpc_batch_limit): rpc_url, rpc_timeout, rpc_batch_limit):
# setproctitle('Block loader: worker %s' % name) setproctitle('Block loader: worker %s' % name)
self.rpc_url = rpc_url self.rpc_url = rpc_url
self.rpc_timeout = rpc_timeout self.rpc_timeout = rpc_timeout
self.rpc_batch_limit = rpc_batch_limit self.rpc_batch_limit = rpc_batch_limit
@ -221,8 +209,8 @@ class Worker:
self.loop.set_default_executor(ThreadPoolExecutor(20)) self.loop.set_default_executor(ThreadPoolExecutor(20))
self.out_writer = out_writer self.out_writer = out_writer
self.in_reader = in_reader self.in_reader = in_reader
self.coins = MRU(100000) # self.coins = MRU(100000)
self.destroyed_coins = MRU(100000) # self.destroyed_coins = MRU(100000)
signal.signal(signal.SIGTERM, self.terminate) signal.signal(signal.SIGTERM, self.terminate)
self.loop.create_task(self.message_loop()) self.loop.create_task(self.message_loop())
self.loop.run_forever() self.loop.run_forever()
@ -257,10 +245,10 @@ class Worker:
inp = block["rawTx"][z]["vIn"][i] inp = block["rawTx"][z]["vIn"][i]
outpoint = b"".join((inp["txId"], int_to_bytes(inp["vOut"]))) outpoint = b"".join((inp["txId"], int_to_bytes(inp["vOut"])))
try: try:
r = self.coins.delete(outpoint) # r = self.coins.delete(outpoint)
block["rawTx"][z]["vIn"][i]["_c_"] = r # block["rawTx"][z]["vIn"][i]["_c_"] = r
t += 1 t += 1
self.destroyed_coins[r[0]] = True # self.destroyed_coins[r[0]] = True
except: except:
pass pass
for i in block["rawTx"][z]["vOut"]: for i in block["rawTx"][z]["vOut"]:
@ -271,7 +259,7 @@ class Worker:
except: except:
address = b"".join((bytes([block["rawTx"][z]["vOut"][i]["nType"]]), address = b"".join((bytes([block["rawTx"][z]["vOut"][i]["nType"]]),
block["rawTx"][z]["vOut"][i]["addressHash"])) block["rawTx"][z]["vOut"][i]["addressHash"]))
self.coins[o] = (pointer, block["rawTx"][z]["vOut"][i]["value"], address) # self.coins[o] = (pointer, block["rawTx"][z]["vOut"][i]["value"], address)
blocks[x] = block blocks[x] = block
if blocks: if blocks:
blocks[x]["checkpoint"] = x blocks[x]["checkpoint"] = x
@ -280,7 +268,7 @@ class Worker:
for i in blocks[x]["rawTx"][y]["vOut"]: for i in blocks[x]["rawTx"][y]["vOut"]:
try: try:
pointer = (x << 42) + (y << 21) + i pointer = (x << 42) + (y << 21) + i
r = self.destroyed_coins.delete(pointer) # r = self.destroyed_coins.delete(pointer)
blocks[x]["rawTx"][y]["vOut"][i]["_s_"] = r blocks[x]["rawTx"][y]["vOut"][i]["_s_"] = r
except: pass except: pass
@ -298,7 +286,6 @@ class Worker:
while True: while True:
msg_type, msg = await self.pipe_get_msg(self.reader) msg_type, msg = await self.pipe_get_msg(self.reader)
if msg_type == b'pipe_read_error': if msg_type == b'pipe_read_error':
self.log.critical("pipe_read_error")
return return
if msg_type == b'get': if msg_type == b'get':