From 4c2bf6873a5349d40ad0e859041db7d2d8293de3 Mon Sep 17 00:00:00 2001 From: Stuckinaboot Date: Wed, 8 May 2019 19:48:21 -0400 Subject: [PATCH] Move bee movie demo to examples and cleanup --- {demos => examples}/bee_movie/bee_movie.py | 0 .../bee_movie/msg_ordering_node.py | 19 ------------------- .../bee_movie/ordered_queue.py | 17 ++--------------- .../bee_movie/test_bee_movie.py | 9 ++++----- 4 files changed, 6 insertions(+), 39 deletions(-) rename {demos => examples}/bee_movie/bee_movie.py (100%) rename {demos => examples}/bee_movie/msg_ordering_node.py (74%) rename {demos => examples}/bee_movie/ordered_queue.py (67%) rename {demos => examples}/bee_movie/test_bee_movie.py (98%) diff --git a/demos/bee_movie/bee_movie.py b/examples/bee_movie/bee_movie.py similarity index 100% rename from demos/bee_movie/bee_movie.py rename to examples/bee_movie/bee_movie.py diff --git a/demos/bee_movie/msg_ordering_node.py b/examples/bee_movie/msg_ordering_node.py similarity index 74% rename from demos/bee_movie/msg_ordering_node.py rename to examples/bee_movie/msg_ordering_node.py index dbe8b9c..e0fa82a 100644 --- a/demos/bee_movie/msg_ordering_node.py +++ b/examples/bee_movie/msg_ordering_node.py @@ -10,19 +10,8 @@ from ordered_queue import OrderedQueue SUPPORTED_PUBSUB_PROTOCOLS = ["/floodsub/1.0.0"] BEE_MOVIE_TOPIC = "bee_movie" -# Message format: -# Sending crypto: ,, -# Ex. send,aspyn,alex,5 -# Set crypto: , -# Ex. set,rob,5 -# Determine message type by looking at first item before first comma class MsgOrderingNode(): - """ - Node which has an internal balance mapping, meant to serve as - a dummy crypto blockchain. There is no actual blockchain, just a simple - map indicating how much crypto each user in the mappings holds - """ def __init__(self): self.balances = {} @@ -71,12 +60,10 @@ class MsgOrderingNode(): asyncio.ensure_future(self.handle_incoming_msgs()) async def publish_bee_movie_word(self, word, msg_id=None): - # print("Publish hit for " + word) my_id = str(self.libp2p_node.get_id()) if msg_id is None: msg_id = self.next_msg_id_func() packet = generate_RPC_packet(my_id, [BEE_MOVIE_TOPIC], word, msg_id) - # print("Packet generated") await self.floodsub.publish(my_id, packet.SerializeToString()) async def handle_bee_movie_word(self, seqno, word): @@ -87,9 +74,3 @@ class MsgOrderingNode(): # Get just the word (and not the seqno) and return the word next_word = (await self.priority_queue.get())[1] return next_word - # if (self.priority_queue.qsize() > 0 and \ - # self.priority_queue[0][0] == self.last_word_gotten_seqno + 1): - # return await self.priority_queue.get() - # else: - # # Wait until a word is put with next_word_seqno == self.last_word_gotten_seqno + 1 - # # Then return first element of priority queue diff --git a/demos/bee_movie/ordered_queue.py b/examples/bee_movie/ordered_queue.py similarity index 67% rename from demos/bee_movie/ordered_queue.py rename to examples/bee_movie/ordered_queue.py index a540297..831744d 100644 --- a/demos/bee_movie/ordered_queue.py +++ b/examples/bee_movie/ordered_queue.py @@ -12,13 +12,10 @@ class OrderedQueue(): self.task = None async def put(self, item): - # print("Put " + str(item)) seqno = item[0] await self.queue.put(item) - # print("Added to queue " + str(item)) if self.last_gotten_seqno + 1 == seqno and self.task is not None: # Allow get future to return the most recent item that is put - # print("Set called") self.task.set() async def get(self): @@ -26,30 +23,20 @@ class OrderedQueue(): front_item = await self.queue.get() if front_item[0] == self.last_gotten_seqno + 1: - # print("Trivial get " + str(front_item)) self.last_gotten_seqno += 1 return front_item else: # Put element back as it should not be delivered yet - # print("Front item put back 1") - # print(str(self.last_gotten_seqno)) - # print(front_item[1]) await self.queue.put(front_item) - # print("Front item put back 2") - # print("get 2") - # Wait until + # Wait until item with subsequent seqno is put on queue self.task = asyncio.Event() - # print("get 3") await self.task.wait() - # print("get 4") item = await self.queue.get() - # print("get 5") - # print(str(item) + " got from queue") # Remove task self.task = None self.last_gotten_seqno += 1 - return item \ No newline at end of file + return item diff --git a/demos/bee_movie/test_bee_movie.py b/examples/bee_movie/test_bee_movie.py similarity index 98% rename from demos/bee_movie/test_bee_movie.py rename to examples/bee_movie/test_bee_movie.py index a305aa3..14b6dfe 100644 --- a/demos/bee_movie/test_bee_movie.py +++ b/examples/bee_movie/test_bee_movie.py @@ -27,12 +27,11 @@ def create_setup_in_new_thread_func(dummy_node): async def perform_test(num_nodes, adjacency_map, action_func, assertion_func): """ - Helper function to allow for easy construction of custom tests for dummy account nodes + Helper function to allow for easy construction of custom tests for msg ordering nodes in various network topologies :param num_nodes: number of nodes in the test :param adjacency_map: adjacency map defining each node and its list of neighbors - :param action_func: function to execute that includes actions by the nodes, - such as send crypto and set crypto + :param action_func: function to execute that includes actions by the nodes :param assertion_func: assertions for testing the results of the actions are correct """ @@ -54,12 +53,12 @@ async def perform_test(num_nodes, adjacency_map, action_func, assertion_func): # Start a thread for each node so that each node can listen and respond # to messages on its own thread, which will avoid waiting indefinitely # on the main thread. On this thread, call the setup func for the node, - # which subscribes the node to the CRYPTO_TOPIC topic + # which subscribes the node to the BEE_MOVIE_TOPIC topic for dummy_node in dummy_nodes: thread = Thread(target=create_setup_in_new_thread_func(dummy_node)) thread.run() - # Allow time for nodes to subscribe to CRYPTO_TOPIC topic + # Allow time for nodes to subscribe to BEE_MOVIE_TOPIC topic await asyncio.sleep(0.25) # Perform action function