1
\$\begingroup\$

I'm new to programming in Python, and I have a task to try and create continuous queue but with delay support. I need to use this for a messaging queue and simulate network latency at different rates over a single session between a server and a client. It feels like what I've done is overkill or way too much architecture. I'd love to have someone look at it and see what I could improve to tighten this up.

I need to:

  • Process the items in order
  • Keep the items processing in a timely manner so tasks aren't held up by work being done
  • Keep it running for new messages in the future without having to run it over again

Here's what I wrote:

import multiprocessing, requests, threading
import decimal
import random
import copy
import time
from enum import Enum
from queue import Queue
from queue import Empty

from multiprocessing.dummy import Pool as ThreadPool
from threading import Thread
from datetime import datetime, timedelta

class WorkQueue:
    def __init__(self, threads=6):
        self.threads = threads

    def process(self, work):

        pool = ThreadPool(self.threads)
        results = pool.map(DelayedTask.process, work)
        pool.close()
        pool.join()




class DelayedTask:
    def __init__(self, func, delay, message):
        print("DelayTask.__init__: {0}".format((func.__name__, delay, message)))
        self.func = func
        self.time = datetime.now() + timedelta(milliseconds=delay)
        self.message = message

    def process(self):

        delta =  self.time - datetime.now()
        if delta.total_seconds() > 0.01:
            print('DelayTask.Process: Sleeping {0} milliseconds\n'.format(round(delta.total_seconds() * 1000)))
            time.sleep(delta.total_seconds())
            self.func(self.message)
        elif delta.total_seconds() < 0.01 and delta.total_seconds() > 0:
            print('DelayTask.Process: Processing with {0} milliseconds remaining\n'.format(round(delta.total_seconds() * 1000)))
            self.func(self.message)
        else:
            print("DelayTask.Process: Processing task: {0} milliseconds late\n".format(round(delta.total_seconds() * -1000)))
            self.func(self.message)
        return True

    def __str__(self):
        return str((self.func.__name__, self.time, self.message))

def get(url):

    #print("Requesting {0}".format(url))

    r = requests.get(url=url)
    print("get(url): Received response for {0} with Status Code {1}".format(url, r.status_code))

aggregatorq = multiprocessing.Queue()

def collector():
    bucket = []
    while len(bucket) <= 10:
        task = aggregatorq.get()
        print("collector: aggregating Tasks\n")
        bucket.append(DelayedTask(task['func'], task['delay'], task['message']))

        if(len(bucket) == 10):
            bucket.sort(key=lambda x: x.time, reverse=False)
            firsttask = bucket[0]
            firsttime =  firsttask.time - datetime.now()
            if firsttime.total_seconds() >= 0:
                print('collector: Sleeping {0} seconds until first task in bucket\n'.format(firsttime.total_seconds()))
                time.sleep(firsttime.total_seconds())
            queue = WorkQueue(10)
            queue.process(bucket)
            bucket = []

def controller():
    print("Starting Controller\n")
    finishtime = datetime.now() + timedelta(seconds=5)
    print("controller: Will finish at {0}\n".format(finishtime))

    sites = ["att", "google", "netflix", "msn", "yahoo", "gmail"]
    while True:
        if datetime.now() > finishtime:
            #print("Controller Finished")
            break;

        requestdelay = random.randint(1, 20)
        randomsite = random.randint(0, len(sites)-1)
        aggregatorq.put({'func': get, 'delay': requestdelay, 'message': 'http://www.{}.com'.format(sites[randomsite])})

t = threading.Thread(target=controller)
t2 = threading.Thread(target=collector)

t.start()
t2.start()

Output:

Starting Controller
collector: aggregating Tasks
controller: Will finish at 2018-01-20 23:55:48.348425
DelayTask.__init__: ('get', 11, 'http://www.netflix.com')
collector: aggregating Tasks
DelayTask.__init__: ('get', 1, 'http://www.gmail.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 11, 'http://www.att.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 5, 'http://www.att.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 8, 'http://www.att.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 12, 'http://www.google.com')
collector: aggregating Tasks

DelayTask.Process: Processing task: 113 milliseconds late
DelayTask.Process: Processing task: 104 milliseconds late
DelayTask.Process: Processing task: 105 milliseconds late
DelayTask.Process: Processing task: 102 milliseconds late
DelayTask.Process: Processing task: 100 milliseconds late
DelayTask.Process: Processing task: 97 milliseconds late
DelayTask.Process: Processing task: 84 milliseconds late
DelayTask.Process: Processing task: 78 milliseconds late
DelayTask.Process: Processing task: 86 milliseconds late
DelayTask.Process: Processing task: 73 milliseconds late
DelayTask.__init__: ('get', 14, 'http://www.netflix.com')

collector: aggregating Tasks

DelayTask.__init__: ('get', 2, 'http://www.gmail.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 11, 'http://www.yahoo.com')get(url): Received response for http://www.google.com with Status Code 200
collector: aggregating Tasks
DelayTask.__init__: ('get', 2, 'http://www.att.com')
get(url): Received response for http://www.netflix.com with Status Code 403get(url): Received response for http://www.netflix.com with Status Code 403

get(url): Received response for http://www.att.com with Status Code 200
get(url): Received response for http://www.att.com with Status Code 200
get(url): Received response for http://www.att.com with Status Code 200
get(url): Received response for http://www.att.com with Status Code 200
get(url): Received response for http://www.yahoo.com with Status Code 200
get(url): Received response for http://www.gmail.com with Status Code 200
get(url): Received response for http://www.gmail.com with Status Code 200
collector: aggregating Tasks

DelayTask.__init__: ('get', 9, 'http://www.att.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 19, 'http://www.google.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 15, 'http://www.netflix.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 15, 'http://www.yahoo.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 13, 'http://www.yahoo.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 3, 'http://www.netflix.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 15, 'http://www.netflix.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 8, 'http://www.gmail.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 4, 'http://www.gmail.com')
collector: aggregating Tasks

DelayTask.__init__: ('get', 10, 'http://www.gmail.com')
collector: Sleeping 0.00141 seconds until first task in bucket
DelayTask.Process: Processing with 1 milliseconds remaining
DelayTask.Process: Processing with 6 milliseconds remaining
DelayTask.Process: Processing task: 2 milliseconds late
DelayTask.Process: Processing task: 0 milliseconds late
DelayTask.Process: Processing with 6 milliseconds remaining
DelayTask.Process: Processing with 4 milliseconds remaining
DelayTask.Process: Processing with 5 milliseconds remaining
DelayTask.Process: Processing with 8 milliseconds remaining
DelayTask.Process: Processing task: 4 milliseconds late
DelayTask.Process: Processing with 7 milliseconds remaining
Improve this question
\$\endgroup\$
5
  • \$\begingroup\$ The biggest issue I'm facing is how to incorporate threading around this example so my tasks won't be held up by current work. Any help would be appreciated. \$\endgroup\$
    – Bryan Davis
    Jan 19, 2018 at 23:46
  • 3
    \$\begingroup\$ If I understand correctly, your code doesn't work as you want to just yet, is that correct? \$\endgroup\$
    – Phrancis
    Jan 20, 2018 at 0:48
  • \$\begingroup\$ Yea, I have a small update where I added a second layer with a ThreadPool but: A. I think I'm a newbie and it's much more code than is necessary. B. I can't figure out how to make it continuous so new adds trigger the process to start up again. I'm using this to simulate latency in messages at various rates over a single session so it needs to stay up and delay messages as the simulated latency changes. \$\endgroup\$
    – Bryan Davis
    Jan 20, 2018 at 2:08
  • \$\begingroup\$ I updated the question to incorporate the ThreadPool but it's still not working exactly right and I still can't figure out how to make it continuous. \$\endgroup\$
    – Bryan Davis
    Jan 20, 2018 at 2:18
  • 1
    \$\begingroup\$ This would be off-topic for this site at the time being, although you're welcome to post it again for improvement once you get it working right. You might try Stack Overflow I think it would be a good fit there to get help with your problem(s), and I've edited the formatting in your question to present it so you can just copy & paste on that site (press the "edit" link to get the raw text) \$\endgroup\$
    – Phrancis
    Jan 20, 2018 at 2:35

0

Reset to default

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.