1
\$\begingroup\$
import threading
import os
import sys
import time
l=0
length=[]
def File_Opener(keys,l):
    for lines in keys:
        l = l + len(lines)-1
    length.append(l)
    l=0
if __name__ == "__main__":
    start=time.time()
    f=open(sys.argv[1],"r")
    print "File to be anaysed is %s" %sys.argv[1]
    Threads=[]
    keys=[]
    for i in range(100):
        try:
            for j in range(200000):
                keys.append(f.next())
        except:
            pass
        t = threading.Thread(target=File_Opener,args=(keys,l))
        t.start()
        keys=[]
        print "Created thread %d" %i
        Threads.append(t)
    for thread in Threads:
        thread.join()
print "Total Length of words are", sum(length)
print "Time taken is %f" %(time.time()-start)
print "Done!!"

Issue : Takes 5x more time than non-multi threaded code.

I am trying to get sum of all chars in a file of very big size. The reason why i used the for loop is to divide the file into small list and pass each to a thread and except is used for in case the file does't come complete in 200000 chars I want to know as to how it can be done with multi threading also. Doing it in python2.7 in Linux

\$\endgroup\$
  • \$\begingroup\$ Are you trying to get the sum in the file, or have you already accomplished that and you are just looking for an optimal way? \$\endgroup\$ – dfhwze Jun 4 at 8:41
  • \$\begingroup\$ I have accomplished it but the time is the issue \$\endgroup\$ – Divyanshu Jun 4 at 8:59
  • \$\begingroup\$ Please clarify what you're doing. What are l, length, lines and keys? As far as I can tell, you're iterating over the characters in File_Opener, not the lines. Are you counting lines, counting characters, or are you summing line count or summing character count? Why 100 and 200000? Why do you need multithreading to do this? \$\endgroup\$ – Daniel Jun 4 at 12:04
3
\$\begingroup\$

Before starting to talk about threads in Python, let's compare to a sequential implementation. I’ll start by building a test file using the following code:

import string
import random

with open('test.txt', 'w') as f:
    for _ in range(20000000):
        line = ''.join(random.choice(string.printable) for _ in range(random.randint(0, 10000)))
        print(line, file=f)

This is Python 3 syntax, and I could not recommend more to use it instead of the end-of-life Python 2.7.

Anyway, after a while, I have a 5.9G file of 64 or so million lines (yes, string.printable contains the newline character) so I stopped it. Let's read it and count the length of its lines.


For starter (and you may have noticed in the snippet above), you should use the with statement to open files so you don't accidentally forget to close them. You can also iterate directly over the content of the file instead of using next; and if you want to know the line number (to spawn a thread each 200000 lines, for instance), you can always delegate the task to enumerate. This way, you’ll avoid using range instead of xrange and consume memory where you don't need to.

Second, timing should be done using timeit for better precision and less boilerplate. Let's start by timing the reading of the file only to get a ballpark figure of how much time we can expect to gain. We’ll use part of the consume recipe from itertools:

from collections import deque


def read_file(filename):
    with open(filename) as f:
        deque(f, maxlen=0)


if __name__ == '__main__':
    import timeit
    time = timeit.timeit('read_file("test.txt")', 'from __main__ import read_file', number=1)
    print('execution took {}s'.format(time))

Result is execution took 41.8203251362s on my machine. This is roughly as fast as I can expect the file to be read… without taking caching into account. If we change the timing part to min(timeit.repeat('read_file("test.txt")', 'from __main__ import read_file', number=1, repeat=5)) we get somewhere around 3.5s.


Now, on to counting characters. As I read it, you want to remove the newline character that is found at the end of each line. You can, as you do here, subtract 1 from each line length; or, alternatively, sum each line length and subtract the amount of lines afterwards. The second approach is easier as we can use map to produce a list of all line lengths and then, sum it before subtracting its len:

def count_characters(filename):
    with open(filename) as f:
        line_lengths = map(len, f)
    return sum(line_lengths) - len(line_lengths)

Timings approaches the same 40s figure for a single execution, but is about 6.5s on repeated executions. This is only 3 seconds of counting characters in 64 million lines; that's somewhat acceptable, but map is building a list in memory which is 64 million items long, so maybe we could speed things up if we didn't:

def count_characters(filename):
    count = 0
    with open(filename) as f:
        for line in f:
            count += len(line) - 1
    return count

Unfortunately, a for loop in Python is slower than a for loop in C (which is what map does) so this code is not faster (albeit being roughly as fast).


Now, trying to speed things up. To parallelize tasks on chunk of data, we usually use map from multiprocessing.Pool. But serializing this much data to feed other processes takes time and isn't worth it. So we are left trying to use threading as you did. However, due to the GIL, you will not be able to run more than one thread in parallel and will be either reading the file or summing the lines; exactly as it is done in sequential code, but you’ll have more overheads due to using threads and helper data structures to move data around. So all in all, for this kind of problem, you’re better off sticking to the sequential implementation.

But let's analyze your code anyway:

  • The l parameter is useless, since it's an integer, it is immutable and the local version on each thread won't interfere with others so passing it as a parameter with seemingly a value of 0 and then resetting it to 0 after the computation is useless;
  • Contrarily, the length list as a global variable is more problematic, I’d rather pass it as a parameter, leading to the threaded functions being:

    def count_characters_in_chunk(lines, accumulator):
        length = sum(len(line) - 1 for line in lines)
        accumulator.append(length)
    
  • As said previously, organizing lines into chunks without prior knowledge of its length can be done using enumerate but we need to account for the last chunk not being the full size, if any:

    def count_characters_in_file(filename, chunk_size=200000):
        threads = []
        lengths = []
    
        with open(filename) as f:
            lines = []
            for line_number, line in enumerate(f, start=1):
                lines.append(line)
                if line_number % chunk_size == 0:
                    t = threading.Thread(target=count_characters, args=(lines, lengths))
                    t.start()
                    threads.append(t)
                    lines = []
    
            if lines:
                t = threading.Thread(target=count_characters, args=(lines, lengths))
                t.start()
                threads.append(t)
    
        for t in threads:
            t.join()
        return sum(lengths)
    

However, there is still two issues with this code:

  1. The repetition of thread management is ugly and error-prone, plus the manual handling of the lines list is unnecessarily verbose. You can use itertools.islice to simplify all that;
  2. The use of a simple list to store resulting lengths is a bad habit to have as threaded code is prone to race-conditions (although highly unlikely in this case) that can lead to loss of data. You should use a Queue.Queue instead.

Final code being:

import itertools
import threading
import Queue


def extract_from_queue(queue):
    while not queue.empty():
        yield queue.get()


def count_characters(lines, accumulator):
    length = sum(len(line) - 1 for line in lines)
    accumulator.put(length)


def count_characters_in_file(filename, chunk_size=200000):
    threads = []
    lengths = Queue.Queue()

    with open(filename) as f:
        while True:
            lines = list(itertools.islice(f, chunk_size))
            if not lines:
                break
            t = threading.Thread(target=count_characters, args=(lines, lengths))
            t.start()
            threads.append(t)

    for t in threads:
        t.join()
    return sum(extract_from_queue(lengths))


if __name__ == '__main__':
    import sys
    count_characters_in_file(sys.argv[1])

But timings using caching indicates 44 seconds on my machine, so not really worth it given the speed and simplicity of the sequential implementation.

\$\endgroup\$
  • \$\begingroup\$ Is the threading implementation done by me correct and is there any possible improvement in the threading script? I actually want to demonstrate the time status for operation of file analysis when done using multi threading multiprocessing and multiplexing. Is the multi threading version of mine accurate enough. As you mentioned to change the for loop with enumerate that would load the full file first but then diving that file into small parts for each threads would be more expensive. \$\endgroup\$ – Divyanshu Jun 5 at 7:07
  • \$\begingroup\$ @Divyanshu Please see updated answer. \$\endgroup\$ – 409_Conflict Jun 5 at 9:29

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.