Why aren't my consumers consuming?
Question: **EDIT: I have tracked down the problem to be the part of my program where I
download the Zip file and parse it. If I comment that out and replace it with
a default line, it parses 10,000 times with no problem.**

Not sure how much of this question should be edited to reflect that finding.

I've written a python program which downloads a zip file which contains a
single log file of about 10,000 lines. It then parses this file, line by line,
and puts the data into a database.

Eventually my script will run through 200 servers/zip files and process about
100,000 lines. (Not all servers have the needed file)

However, currently when I run the script with 1 consumer, I only get about 13
lines processed into the database. If I run 2 consumers, I get 24. If I run 10
consumers I get 100, and if I run 20 consumers I get 240.

Sometimes, the result of running the script is "Consumer Finished" with said
number of entries in the database (far short of the 10K-30K I'm expecting) But
other times, I get an error message:

    
    
    > Traceback (most recent call last):   File
    > "C:\Python27\lib\multiprocessing\queues.py", line 262, in _feed
    >     send(obj) IOError: [Errno 232] The pipe is being closed
    

What can be causing this problem? Attached is a modified version of my code to
remove sensitive data:

    
    
    import urllib, urlparse
    import sys
    import os
    import datetime
    from calculon import Calculon
    import random
    import pprint
    import time
    import random
    import urllib, urlparse
    import traceback
    import psycopg2
    import psycopg2.extras
    from datetime import date, datetime, time, timedelta
    import os.path
    import requests
    import io
    import urllib2, cStringIO, zipfile
    import re
    import httplib
    import urlparse
    
    def daterange(start_date, end_date):
      for n in range(int((end_date - start_date).days)):
        yield start_date + timedelta(n)
    
    
    
    def producer(args):
      print "Producing!"
      logdt_start = args["logdt_start"]
      logdt_end = args["logdt_end"]
      for single_date in daterange(logdt_start, logdt_end):
        logdt = single_date + timedelta(days=1)
        print "Reading log file..."
        for x in range(1,2):
          servername = "server-{0}".format("%02d" % (x,))
          filename = "zipped_log.log{0}".format(logdt.date().isoformat())
          url = "http://url.to.zip.file/{0}/{1}".format(servername, filename)
          zip_path = 'path/to/file/within/zip/{0}/{1}'.format(servername, filename)
          if httpExists(url):
            try:
               request = urllib2.urlopen(url)
               zipinmemory =  cStringIO.StringIO(request.read())
                 with zipfile.ZipFile(zipinmemory) as archive:
                   with archive.open(zip_path) as log:
                     print "File Found! Reading %s..." % filename
                     for line in log:
                      args["_queue"].put(line)
    
              print "Queue has approximatly {0} items".format(args["_queue"].qsize())
    
            except:
              print "exception could not load %s" % url
              traceback.print_exc()
        return True
    
    
    def httpExists(url):
        host, path = urlparse.urlsplit(url)[1:3]
        found = 0
        try:
            connection = httplib.HTTPConnection(host)  ## Make HTTPConnection Object
            connection.request("HEAD", path)
            responseOb = connection.getresponse()      ## Grab HTTPResponse Object
    
            if responseOb.status == 200:
                found = 1
            #else:
                #print "Status %d %s : %s" % (responseOb.status, responseOb.reason, url)
        except Exception, e:
            print e.__class__,  e, url
        return found
    
    
    def parse_log(line):
        if len(line) < 10 or line[0] != '[':
            return {}
        mod_line = line
        mod_line = mod_line.replace('  ', ' ') #whats this for?
        query_dict = {}
        match = re.search('([\d:\/\s]+)\sUTC', mod_line)
        s = match.start()
        e = match.end() - 5
        query_dict['date_ts'] = datetime.strptime(mod_line[s:e], '%d/%m/%Y %H:%M:%S:%f')
        e = e+2
        mod_line = mod_line[e:]
        match = re.search('(\w+)\sLogger:\s', mod_line)
        e = match.end()
        query_dict['status'] = match.group(1)
        mod_line = mod_line[e:]
        for key_value in re.split(',', mod_line):
            keypair = re.search('(\w+)=(\w+)', key_value)
            key = keypair.group(1)
            value = keypair.group(2)
            query_dict[key] = value
        return query_dict
    
    def consumer(args):
      global consumed
      consumed += 1
      print "Consumed : {0}".format(consumed)
      try:
        db = args["db"]
        cname = args["cname"]
        arg_value = args["_value"]
    
        cur = db.cursor()
        error_count = 0
    
    
        if arg_value is None:
            print "Consumer Finished!"
            return False
        line = arg_value
        qdict = parse_log(line)
    
        if len(qdict) == 0:
            print "No data to consumer %s" % cname
            return None
    
        query = """
        INSERT INTO my_db(date_ts,
            status, cmd, creativeString, environment_id, client_type_id, platform_id, sn_type_id, user_id,
            device_id, invoker_sn_id, invoker_type, poster_sn_id, origin, event_type, creative_id, ch,
             src, afp, cmp, p1, p2,p3)
        VALUES (%(date_ts)s,%(status)s,%(cmd)s,%(creativeString)s,%(environment_id)s,%(client_type_id)s,%(platform_id)s,
        %(sn_type_id)s,%(user_id)s,%(device_id)s,%(invoker_sn_id)s,%(invoker_type)s,%(poster_sn_id)s,%(origin)s,
        %(event_type)s,%(creative_id)s,%(ch)s, %(src)s, %(afp)s, %(cmp)s,
        %(p1)s, %(p2)s, %(p3)s);
        """
    
        try:
          cur.execute(cur.mogrify(query, qdict))
          db.commit()
          global processed
          processed += 1
          print "processed : {0}".format(processed)
        except:
          error_count = error_count + 1
          print "ERROR in insert {0}".format(error_count)
          traceback.print_exc()
          print qdict
          sys.exit(2)   
      except:
          print "Error in parsing:  "  + val
          tracback.print_exc()
          sys.exit(12)
    
    def main():
      log_start = datetime(2015,1,19);
      log_end = datetime(2015,1,20);
      consumer_args_list = []
    
      noOfConsumers = 1;
      for x in range(0, noOfConsumers):
        print "Creating Consumer {0}".format(x)
        print "Connecting to logs db..."
        db_name = 'mydb'
        connString = "dbname={0} host={1} port={2} user={3} password={4}".format(db_name, 'localhost', 5433, 'postgres',                                                                       'pword')
        db = psycopg2.connect(connString)
        consumer_args = {"cname": "CONSUMER_{0}".format(x), "db":db}
        consumer_args_list.append(consumer_args)
    
      calculon = Calculon( producer, 
            [{"logdt_start": log_start,
              "logdt_end": log_end}],
                True,
                consumer,
                consumer_args_list,
                True)
      result = calculon.start()
    
    consumed = 0
    processed = 0
    if __name__ == "__main__":
      main()
    

The output looks like this:

    
    
    > Creating Consumer 0
    Connecting to logs db...
    Producing!
    Reading log file...
    File Found! Reading log2015-01-20...
    Queue has approximatly 9549 items
    Consumed : 1
    processed : 1
    Consumed : 2
    processed : 2
    Consumed : 3
    processed : 3
    Consumed : 4
    processed : 4
    Consumed : 5
    processed : 5
    Consumed : 6
    processed : 6
    Consumed : 7
    processed : 7
    Consumed : 8
    processed : 8
    Consumed : 9
    processed : 9
    Consumed : 10
    processed : 10
    Consumed : 11
    processed : 11
    Consumed : 12
    processed : 12
    Consumed : 13
    Traceback (most recent call last):
      File "C:\Python27\lib\multiprocessing\queues.py", line 262, in _feed
        send(obj)
    IOError: [Errno 232] The pipe is being closed
    


Answer: The error turned out to be a bad line in the input file, which broke the
regular expression.

For example: One of the values of the comma seperated list was: `foobar=2,
foo=Something here, is ,a really, poor value, bar=2`

I was able to fix the problem by adding the following code within the consumer
method:

    
    
        try:
          qdict = parse_adx_client_log(line)
        except:
          qdict = {}
          print "BAD LINE {0}".format(line)
    
        if len(qdict) == 0:
           print "No data to consumer %s" % cname
           return None
    

