title: Pycassa, threadpool, "Exception in thread Thread-3 (most likely raised during interpreter shutdown):"
body: <p>.I'm trying to speed up an insert into Cassandra using pycassa.
  I heard that using multithreading and opening multiple connections speeds it up a lot. I'm inserting a load of tweets in json format.
  My code here works, for a bit, and then the threads start throwing exceptions and it stops, it seems the more threads I have the faster it stops working... I'm guessing that the problem is the connections to cassandra, something to do with connection pooling. Any ideas?  </p>
  
  <p>EDIT: all threads throw "Exception in thread Thread-3 (most likely raised during interpreter shutdown):"</p>
  
  <pre><code>import time
  import pycassa
  from pycassa.pool import ConnectionPool
  from pycassa.columnfamily import ColumnFamily
  from datetime import datetime 
  import json
  import threadpool
  pool = threadpool.ThreadPool(4)
  kspool = ConnectionPool('TweetsKS',use_threadlocal = True)
  
  def process_tasks(lines):
  
      #let threadpool format your requests into a list
      requests = threadpool.makeRequests(insert_into_cfs, lines)
  
      #insert the requests into the threadpool
      for req in requests:
          pool.putRequest(req) 
  
  
  def read(file):
      bench = open("bench.txt", "w")
      bench.write(str(datetime.now())+"\n")
      """read data from json and insert into keyspace"""
      json_data=open(file)
      lines = []
      for line in json_data:
          lines.append(line)
      process_tasks(lines)
  
  
  def insert_into_cfs(line):
  
      user_tweet_cf = pycassa.ColumnFamily(kspool, 'UserTweet')
      user_name_cf = pycassa.ColumnFamily(kspool, 'UserName')
      tweet_cf = pycassa.ColumnFamily(kspool, 'Tweet')
      user_follower_cf = pycassa.ColumnFamily(kspool, 'UserFollower')
  
      tweet_data = json.loads(line)
      """Format the tweet time as an epoch seconds int value"""
      tweet_time = time.strptime(tweet_data['created_at'],"%a, %d %b %Y %H:%M:%S +0000")
      tweet_time  = int(time.mktime(tweet_time))
  
      new_user_tweet(user_tweet_cf,tweet_data['from_user_id'],tweet_time,tweet_data['id'])
      new_user_name(user_name_cf,tweet_data['from_user_id'],tweet_data['from_user_name'])
      new_tweet(tweet_cf,tweet_data['id'],tweet_data['text'],tweet_data['to_user_id'])
  
      if tweet_data['to_user_id'] != 0:
          new_user_follower(user_follower_cf,tweet_data['from_user_id'],tweet_data['to_user_id'])
  
  
  """4 functions below carry out the inserts into specific column families"""     
  def new_user_tweet(user_tweet_cf,from_user_id,tweet_time,id):
      user_tweet_cf.insert(from_user_id,{(tweet_time): id})
  
  def new_user_name(user_name_cf,from_user_id,user_name):
      user_name_cf.insert(from_user_id,{'username': user_name})
  
  def new_tweet(tweet_cf,id,text,to_user_id):
      tweet_cf.insert(id,{
      'text': text
      ,'to_user_id': to_user_id
      })  
  
  def new_user_follower(user_follower_cf,from_user_id,to_user_id):
      user_follower_cf.insert(from_user_id,{to_user_id: 0})   
  
  if __name__ == '__main__':
      read('tweets.json')
  </code></pre>