StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add end messages into simplified logic

Browse Source
This commit is contained in:
Stuckinaboot 2019-04-07 18:40:01 -04:00
parent 470f5e6e51
commit 9277be98bf
2 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/sharding/driver.py
View File

@ -99,7 +99,7 @@ async def main():
num_receivers_in_each_topic[topic] = len(topic_map[topic]) num_receivers_in_each_topic[topic] = len(topic_map[topic])
print("Performing test") print("Performing test")
await sender.perform_test(num_receivers_in_each_topic, topics, 1) await sender.perform_test(num_receivers_in_each_topic, topics, 1)
print("All testing completed")
await cleanup() await cleanup()

24
examples/sharding/sender.py
View File

@ -45,18 +45,17 @@ class SenderNode():
self.test_being_performed = True self.test_being_performed = True
this = self self.all_streams = []
all_streams = []
async def ack_stream_handler(stream): async def ack_stream_handler(stream):
all_streams.append(stream) self.all_streams.append(stream)
while self.test_being_performed: while self.test_being_performed:
# This Ack is what times out when multi-topic tests finish # This Ack is what times out when multi-topic tests finish
ack = await stream.read() ack = await stream.read()
if ack is not None: if ack is not None:
await self.ack_queue.put(ack) await self.ack_queue.put(ack)
else: else:
print("FUCK")
break break
# Reached once test_being_performed is False # Reached once test_being_performed is False
# Notify receivers test is over # Notify receivers test is over
@ -101,20 +100,25 @@ class SenderNode():
curr_time = timer() curr_time = timer()
async def end_all_async(): async def end_all_async():
# Add None to all queues indicating that we should break the loop # Add None to ack_queue to break out of the loop.
# Note: This is necessary given the current code or the code will never
# terminate
await self.ack_queue.put(None) await self.ack_queue.put(None)
for queue in self.topic_ack_queues:
await self.topic_ack_queues[queue].put(None) # This is not necessary but is useful for turning off the receivers gracefully
for stream in self.all_streams:
await stream.write("end".encode())
async def perform_test_on_topic(topic): async def perform_test_on_topic(topic):
print("Performing test on topic " + topic) print("Performing test on topic " + topic)
start = timer() start = timer()
curr_time = timer() curr_time = timer()
# Perform test while time is not up here AND # Perform test while time is not up here AND
# while time is not up in handle_ack_queues, which is checked with the # while time is not up in handle_ack_queues, which is checked with the
# self.test_being_performed boolean # self.test_being_performed boolean
while (curr_time - start) < time_length and self.test_being_performed: while (curr_time - start) < time_length and self.test_being_performed:
# Send message (NOTE THIS IS JUST ONE TOPIC) # Send message on single topic
packet = generate_RPC_packet(my_id, [topic], msg_contents, self.next_msg_id_func()) packet = generate_RPC_packet(my_id, [topic], msg_contents, self.next_msg_id_func())
await self.floodsub.publish(my_id, packet.SerializeToString()) await self.floodsub.publish(my_id, packet.SerializeToString())
@ -137,8 +141,8 @@ class SenderNode():
nonlocal completed_topics_count, num_topics nonlocal completed_topics_count, num_topics
print("Test completed " + topic) print("Test completed " + topic)
completed_topics_count += 1 completed_topics_count += 1
self.test_being_performed = False
if completed_topics_count == num_topics: if completed_topics_count == num_topics:
self.test_being_performed = False
print("End all async") print("End all async")
await end_all_async() await end_all_async()