Application to monitor device power levels for deviations

Question

I have a high performance, computation intensive application that runs on a Centos 7 machine with Python 2.7.5.

I'll try to explain what the application does:

1. The application runs an infinite loop, where it receives a message from an API call representing power levels of a device. The message is in Avro and encoded as a JSON string.
2. Each of the devices gets a maximum of 8 separate power level readings (imagine each being a separate HW component within the device). 8 separate components constitute one device. There are a total of 50 of them. So (50 * 8) power level reports are possible.
3. Each of this 8 HW devices produces a power report once per 30s.
4. I have business logic to compute an arithmetic mean of the first 4 devices (component ids 1-4) and mean for the last 4 devices (ids 5-8).

5. For a given device once I get all 8 readings received, I calculate the above mean and compare the difference between the mean of the group against the individual components, i.e. first 4 - mean_1, last 4 - mean_2.

   for id in 1 2 3 4: do ( mean_1 - pwr_reading(id) )
   for id in 5 6 7 8: do ( mean_2 - pwr_reading(id) )
   

6. If the above difference is below a certain threshold, say thresh_first for first four and thresh_last for last four, I need to do an action.

---

So to model the above requirements, I've created a device class where I'm holding this information

obj_list = {}

class DevPwrInfo(object):
    """ The class provides an abstraction of all the processing we do at one
    device level
    """

    def __init__(self, code):
        """ The constructor spins up a new device object initializing the identifiers
        and the necessary data structures needed for the evaluation
        """
        self.code = code
        self.first4_pwr = {}
        self.last4_pwr = {}
        self.mean_val_first4 = ""
        self.mean_val_last4 = ""
        self.threshold_breach_list_first4 = []
        self.threshold_breach_list_last4 = []


    def reset_dev_info(self):
        """ Clear the data retained after finishing one round of report
        evaluation
        """
        self.first4_pwr = {}
        self.last4_pwr = {}
        self.mean_val_first4 = ""
        self.mean_val_last4 = ""

    def add_dev_pwr(self, id, pwr, pwr_valid_flag):
        if 1 <= int(id) <= 4:
            if pwr_valid_flag:
                self.first4_pwr[id] = pwr
            else:
                self.first4_pwr[id] = 0.0
        else:
            if pwr_valid_flag:
                self.last4_pwr[id] = pwr
            else:
                self.last4_pwr[id] = 0.0

        if len(self.first4_pwr) == 4:
            self.mean_val_first4 = self.compute_mean(first4_pwr)
            self.compare_thresh('first4')
            self.first4_pwr.clear()

        if len(self.last4_pwr) == 4:
            self.last4_pwr = self.compute_mean(last4_pwr)
            self.compare_thresh('last4')
            self.last4_pwr.clear()

    def compute_mean(self, pwr_list):
        return (float(sum(pwr_list)) / max(len(pwr_list), 1))

    def compare_thresh(self, type):
        low_thresh  = thresh_dict[self.code]

        if type == 'first4':
            pwr_dict = self.first4_pwr
            mean_val = self.mean_val_first4
        else:
            pwr_dict = self.last4_pwr
            mean_val = self.mean_val_last4

        for id, pwr in pwr_dict.iteritems():
            if int(math.floor(mean_val - ( pwr ))) < int(low_thresh):
                print("Will add more logic here")

def pwr_report_msg_decode(message):
    """ Handler for the raw idb message from the API
    """
    if message is not None:
        # This API is called for each message from the API call, so that
        # each device's object is called by the string identifier and
        # 'add_dev_pwr' function will ensure the lists are updated
        obj_list[message['code']].add_dev_pwr( message['id'],
                                               message['pwr'],
                                               message['valid_flag'])

# obj_dict is a dict of objects with key name as device name as value as the
# dict object
if __name__ == "__main__":
    # allowed_devices_list contains list of 44 device names
    allowed_devices_list = [ 'abc01', 'def01', 'xyz01' ]
    for device in allowed_devices_list:
        obj_list[device] = DevPwrInfo(device)

    while True:
         # An API producing message in the format
         msg = { "code": "abc01", "id": "3", "pwr": "-59.2", "valid_flag": "True'" }

So my question is how do I make each of the 44 objects run in parallel and not sequentially in one thread. I've looked about ThreadPoolExecutor but not sure how to make it computationally optimum?

Answer 1

I'll also begin with the low hanging fruit mentioned by Mateusz Konieczny. PEP8/pylint/etc. your code. It is decently formatted, but there are some issues. Before even considering optimizing performance, you should first optimize for the person reading your code. Until you've profiled and determined that you need to add complexity (because speed is an issue), programmer productivity (specifically, the ability to quickly glance at your code and understand it) is paramount.

Also, you can often eek out a decent bit of performance by switching to Python 3. Perhaps your hardware prevents this, but it's usually a free performance win. If the math truly is this intensive, running under pypy might also give you a free performance boost.

But, have you profile this code? Do benchmarks indicate that this needs to be optimized? As it exists right now, I find it unlikely that even sequentially this is unable to process 50*8 inputs every 30 seconds (that's 13 ops/sec or 75ms per op, which seems reasonable). If print("Will add more logic here") is computationally intensive, why not just run it in a separate process instead of complicating this relatively simple API request parsing and math?

Running the API requests in parallel could be as simple as using a multiprocessing.Pool (you'll want to use it instead of threads because of the GIL):

with Pool() as pool:
    for msg in api_messages:
        pool.apply(pwr_report_msg_decode, msg)

Although, unfortunately it's not quite that simple. You'd need to make obj_list a shared object (between the processes), which between processes has overhead for writing/read (because you need locks). Also, your API requests may already come in from a threaded context. If you were on python 3, asyncio could probably make expressing this logic a lot easier.

To remedy the locking issue, you may try creating a separate multiprocessing.Process for each of the 50 things. Then you dispatch the API message to the appropriate process via a queue:

queues = [Queue() for _ in range(50)]
processes = [Process(target=handle_thing_readings, args=(queue,))
             for queue in queues]

for msg in api_messages:
    queues[msg['code']].put(msg)

def handle_thing_readings(queue):
    device = DevPwrInfo()
    while True:
        msg = queue.get()
        device.add_dev_pwr(msg['id'], msg['pwr'],
                           msg['valid_flag'])

This does require serializing msg, so you may want to replace the dictionary with a custom object that has __slots__ defined. That said, there is still overhead, but this approach is likely better than locking.

All of the run around here should make it clear such patterns aren't too well suited for Python, especially if performance really is a concern. In my opinion, something like Go is much better suited for a task like this. Thanks to channels and goroutines, you could express all of this complex parallel logic in maybe 10ish lines of Go (and it has some pretty nifty runtime tools for analyzing performance, checking for deadlocks, etc.).