7
\$\begingroup\$

A bit of context: I have 100 addresses that can send transactions to a blockchain, and I have a web app that works like this:

  1. Get request and handle it in a thread
  2. Get a free address from the queue
  3. Send a transaction
  4. Return address to the queue

It's important to know that one address cannot send two transactions at once.

My approach was to make a custom python queue that uses a dict of "address: free/taken" instead of the normal queue.

Its good for two reasons:

  1. I need to be able to get a random address for the dict
  2. The base queue get and put method already handle thread-locking and timeout.

This seems to work, but I would like some feedback on it.

from queue import Queue
from enum import Enum
import random


class ChannelStatuses(Enum):
    FREE = 0
    TAKEN = 1
    UNDERFUNDED = 2


class ChannelDict(Queue):
    def __init__(self, channels_seeds):
        # Init base queue
        super(ChannelDict, self).__init__(len(channels_seeds))
        # Change queue from a 'dqueue' object to a dict full of free channels
        self.queue = {channel: ChannelStatuses.FREE for channel in channels_seeds}

    def _get(self):
        # Get a list of all free channels
        free_channels = self.get_free_channels()
        # Select a random free channel
        selected_channel = random.choice(free_channels)[0]
        # Change channel state to taken
        self.queue[selected_channel] = ChannelStatuses.TAKEN
        return selected_channel

    def _put(self, channel):
        # Change channel state to free
        self.queue[channel] = ChannelStatuses.FREE

    def _qsize(self):
        # Base queue checks if the queue is not empty by checking the length of the queue (_qsize() != 0)
        # We need to check it by checking how many channels are free
        return len(self.get_free_channels())

    def get_free_channels(self):
        # Get a list of channels with "FREE" status
        return list(filter(lambda item: item[1] == ChannelStatuses.FREE, self.queue.items()))

"channel" is another word for "address"

Used this answer as inspiration: https://stackoverflow.com/a/16506527/2126254

More context on how get/put are used,these methods are exposed:

@contextmanager
def get_channel(self, timeout=CHANNEL_QUEUE_TIMEOUT):
    """
    Get the next available channel
    :param float timeout: (Optional) How long to wait before raising an exception
    :return next avilable channel
    :rtype str

    :raises KinErrors.ChannelBusyError
    """
    try:
        channel = self.channels.get(True, timeout=timeout)
    except queue.Empty:
        raise ChannelsBusyError()

    try:
        yield channel
    finally:
        self.put_channel(channel)

def put_channel(self, timeout=CHANNEL_PUT_TIMEOUT):
    """
    Return a channel to the queue
    :param float timeout: (Optional) How long to wait before raising an exception

    :raises KinErrors.ChannelsFullError
    """
    try:
        self.channels.put(True, timeout=timeout)
    except queue.Full:
        raise ChannelsFullError()

And the user will use:

with get_channel() as channel:
    #send transaction
\$\endgroup\$
3
\$\begingroup\$

1. Review

  1. There is no docstring for ChannelDict. The text in the post would make a good start.

  2. A ChannelDict is not a dictionary (it doesn't support the mapping interface, __getitem__ and __setitem__ and so on), so the name is a bit misleading.

  3. ChannelDict is not specialized for channels: it would work for any kind of object. A name like RandomQueue would make it easier for the reader to understand the purpose of the code.

  4. ChannelDict inherits from queue.Queue but has a different interface for initialization (it takes an iterable of channels instead of a maximum size). Perhaps this is convenient for your use case but it makes the code a little harder to understand because you can't just say "it's just like a Queue except that it gets items in random order", you have to explain the difference in initialization and that you can't set a maximum size.

  5. The algorithm for getting a random item from the queue takes time proportional to the number of items, because get_free_channels has to loop over all the items looking for any that are free.

  6. ChannelStatuses.UNDERFUNDED is declared but not used.

2. Revised code

One way to efficiently pick a random item (instead of looping over all the items) is to keep the available items in a list, and to swap the randomly selected item with the last item in the list before popping the selected item.

from queue import Queue
from random import randrange

class RandomQueue(Queue):
    """Variant of Queue that retrieves items in a random order."""
    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        self.queue.append(item)

    def _get(self):
        queue = self.queue
        i = randrange(len(queue))
        queue[i], queue[-1] = queue[-1], queue[i]
        return queue.pop()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.