Python Multithreading Producer Consumer Pattern

Question

I'm still learning how to code and these are my first attempts at multithreading. I've read a bunch of multithreading articles. I thought these were very helpful:

There's quite a lot to think about. Especially for a beginner. Unfortunately, when I try to put this information into practice my code isn't quite working.

The idea behind this code is to read simplified.txt which contains lines of comma delimited numbers. Eg: 0.275,0.28,0.275,0.275,36078. The producer thread reads each line and strips the newline character from the end of the line. Then each number in the line is split and assigned a variable. Variable1 is then placed into the queue. The consumer thread will pick up items in the queue, square it, then add an entry into the log file.

The code I am using comes from this template. This is the code I have so far:

import threading
import queue
import time
import logging
import random
import sys

read_file = 'C:/temp/temp1/simplified.txt'
log1 = open('C:/temp/temp1/simplified_log1.txt', "a+")

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-9s) %(message)s',)

BUF_SIZE = 10
q = queue.Queue(BUF_SIZE)

class ProducerThread(threading.Thread):
    def __init__(self, name, read_file):
        super(ProducerThread,self).__init__()
        self.name = name
        self.read_file = read_file

    def run(self, read_file):
        while True:
            if not q.full():
                with open(read_file, 'r') as f:
                    for line in f:
                        stripped = line.strip('\n\r')
                        value1,value2,value3,value4,value5,value6,value7 = stripped.split(',')
                        q.put(value1)
                        logging.debug('Putting ' + str(value1) + ' : ' + str(q.qsize()) + ' items in queue')
                        time.sleep(random.random())
        return

class ConsumerThread(threading.Thread):
    def __init__(self, name, value1, log1):
        super(ConsumerThread,self).__init__()
        self.name = name
        self.value1 = value1
        self.log1 = log1
        return

    def run(self):
        while True:
            if not q.empty():
                value1 = q.get()
                sqr_value1 = value1 * value1
                log1.write("The square of " + str(value1) + " is " + str(sqr_value1))
                logging.debug('Getting ' + str(value1) + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return

if __name__ == '__main__':
    
    p = ProducerThread(name='producer')
    c = ConsumerThread(name='consumer')

    p.start()
    time.sleep(2)
    c.start()
    time.sleep(2)

When I run the code, I get this error:

Traceback (most recent call last):
  File "c:/Scripta/A_Simplified_Producer_Consumer_Queue_v0.1.py", line 60, in <module>
    p = ProducerThread(name='producer')
TypeError: __init__() missing 1 required positional argument: 'read_file'

I don't know where else I need to add read_file. Any help would be greatly appreciated. Thanks in advance.

score 4 · Answer 1 · answered Feb 02 '19 at 02:39

Your ProducerThread class requires 2 parameters (name and read_file) as arguments to its constructor as defined in its __init__ method, where you only provide the first such argument when you create an instance in your main block. You have the same problem with your second class.

You should either provide the read_file to the constructors when creating instances or just remove it from the constructor signature since you don't appear to use it anyways (you use the read_file passed into run function, but I don't think that is correct). Seems like you're attempting to override that method from the Thread superclass and I doubt that takes such a parameter.

score 1 · Accepted Answer · answered Feb 03 '19 at 03:04

Thank you userSeventeen for setting me on the right path. I thought that in order to use outside variables I needed to place them in the init method, then again into the run method. You've clarified that I only needed to use the variables in the run methods. This is the working code. I had to remove the while true: statement as I did not want the code to run forever.

import threading
import queue
import time
import logging
import random
import sys
import os


read_file = 'C:/temp/temp1/simplified.txt'
log1 = open('C:/temp/temp1/simplified_log1.txt', "a+")

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-9s) %(message)s',)

BUF_SIZE = 10
q = queue.Queue(BUF_SIZE)

class ProducerThread(threading.Thread):
    def __init__(self, name):
        super(ProducerThread,self).__init__()
        self.name = name

    def run(self):
        with open(read_file, 'r') as f:
            for line in f:
                stripped = line.strip('\n\r')
                value1,value2,value3,value4,value5 = stripped.split(',')
                float_value1 = float(value1)
                if not q.full():
                    q.put(float_value1)
                    logging.debug('Putting ' + str(float_value1) + ' : ' + str(q.qsize()) + ' items in queue')
                    time.sleep(random.random())
        return

class ConsumerThread(threading.Thread):
    def __init__(self, name):
        super(ConsumerThread,self).__init__()
        self.name = name
        return

    def run(self):
        while not q.empty():
            float_value1 = q.get()
            sqr_value1 = float_value1 * float_value1
            log1.write("The square of " + str(float_value1) + " is " + str(sqr_value1))
            logging.debug('Getting ' + str(float_value1) + ' : ' + str(q.qsize()) + ' items in queue')
            time.sleep(random.random())
        return

if __name__ == '__main__':

    p = ProducerThread(name='producer')
    c = ConsumerThread(name='consumer')

    p.start()
    time.sleep(2)
    c.start()
    time.sleep(2)

Python Multithreading Producer Consumer Pattern

2 Answers2