Actually use --order argument when running sync jobs

Was not hooked up the the `crawl_blocks_executor` before.

Also removed branching for whether to run `crawl_blocks` or
`crawl_blocks_executor` in the CLI handler. Added `num_processes`
arguments to `crawl_blocks_executor` that callers can use to specify
whether or not to create subprocesses.
pull/44/head
Neeraj Kashyap 2021-08-02 11:46:50 -07:00
rodzic 50cad0f17f
commit 66a9b9a80f
2 zmienionych plików z 32 dodań i 36 usunięć

Wyświetl plik

@ -93,24 +93,18 @@ def ethcrawler_blocks_sync_handler(args: argparse.Namespace) -> None:
) )
time.sleep(20) time.sleep(20)
continue continue
if top_block_number - bottom_block_number >= 10:
for blocks_numbers_list in yield_blocks_numbers_lists( for blocks_numbers_list in yield_blocks_numbers_lists(
f"{bottom_block_number}-{top_block_number}" f"{bottom_block_number}-{top_block_number}",
): order=args.order,
print( ):
f"Adding blocks {blocks_numbers_list[-1]}-{blocks_numbers_list[0]}" print(f"Adding blocks {blocks_numbers_list[-1]}-{blocks_numbers_list[0]}")
) # TODO(kompotkot): Set num_processes argument based on number of blocks to synchronize.
crawl_blocks_executor( crawl_blocks_executor(
block_numbers_list=blocks_numbers_list, block_numbers_list=blocks_numbers_list,
with_transactions=bool(strtobool(args.transactions)),
)
else:
blocks_numbers_list = range(bottom_block_number, top_block_number + 1)
print(f"Adding blocks {bottom_block_number}-{top_block_number - 1}")
crawl_blocks(
blocks_numbers=blocks_numbers_list,
with_transactions=bool(strtobool(args.transactions)), with_transactions=bool(strtobool(args.transactions)),
) )
print(f"Synchronized blocks from {bottom_block_number} to {top_block_number}") print(f"Synchronized blocks from {bottom_block_number} to {top_block_number}")
time.sleep(10) time.sleep(10)

Wyświetl plik

@ -136,6 +136,7 @@ def crawl_blocks_executor(
block_numbers_list: List[int], block_numbers_list: List[int],
with_transactions: bool = False, with_transactions: bool = False,
verbose: bool = False, verbose: bool = False,
num_processes: int = MOONSTREAM_CRAWL_WORKERS,
) -> None: ) -> None:
""" """
Execute crawler in processes. Execute crawler in processes.
@ -153,27 +154,28 @@ def crawl_blocks_executor(
worker_job_lists[i % MOONSTREAM_CRAWL_WORKERS].append(block_number) worker_job_lists[i % MOONSTREAM_CRAWL_WORKERS].append(block_number)
results: List[Future] = [] results: List[Future] = []
if num_processes == 1:
return crawl_blocks(block_numbers_list, with_transactions, verbose)
else:
with ProcessPoolExecutor(max_workers=MOONSTREAM_CRAWL_WORKERS) as executor:
for worker in worker_indices:
if verbose:
print(f"Spawned process for {len(worker_job_lists[worker])} blocks")
result = executor.submit(
crawl_blocks,
worker_job_lists[worker],
with_transactions,
)
result.add_done_callback(record_error)
results.append(result)
with ProcessPoolExecutor(max_workers=MOONSTREAM_CRAWL_WORKERS) as executor: wait(results)
for worker in worker_indices: # TODO(kompotkot): Return list of errors and colors responsible for
if verbose: # handling errors
print(f"Spawned process for {len(worker_job_lists[worker])} blocks") if len(errors) > 0:
result = executor.submit( print("Errors:")
crawl_blocks, for error in errors:
worker_job_lists[worker], print(f"- {error}")
with_transactions,
)
result.add_done_callback(record_error)
results.append(result)
wait(results)
# TODO(kompotkot): Return list of errors and colors responsible for
# handling errors
if len(errors) > 0:
print("Errors:")
for error in errors:
print(f"- {error}")
def process_contract_deployments() -> List[Tuple[str, str]]: def process_contract_deployments() -> List[Tuple[str, str]]:
""" """