poc_checking_heap.py

# -*- coding: utf-8 -*-

import os
import uuid
import time

import cassandra
from cassandra.io.libevreactor import LibevConnection
from cassandra.cluster import Cluster

import concurrent
import os
import gc
import resource
from itertools import cycle
import traceback
import objgraph
from guppy import hpy

from cassandra import ConsistencyLevel


from cassandra.query import SimpleStatement
import sys

connection_class=LibevConnection

import logging

log = logging.getLogger()
log.setLevel('INFO')
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s"))
log.addHandler(handler)




cluster = Cluster(("127.0.0.1", ),  connection_class=connection_class)

session = cluster.connect()
session.execute('''
            CREATE KEYSPACE IF NOT EXISTS test3rf
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}''')

session.execute('''
            CREATE TABLE IF NOT EXISTS test3rf.test (
                k int PRIMARY KEY,
                v int )''')


def read():
    s = "SELECT * FROM test3rf.test"
    session.execute(s)

def write():
    s = "INSERT INTO test3rf.test (k, v) VALUES (1, 1)"
    session.execute(s)

count = 0
hp = hpy()
num_threads = 50

pool = concurrent.futures.ThreadPoolExecutor(max_workers=num_threads)
while 1:
    if not count % 300:
        before = hp.heap()

    for i in range(num_threads/2):
        pool.submit(read)
        pool.submit(write)
    time.sleep(0.0001)

    if not count % 300:
        pool.shutdown(wait=True)
        gc.collect()
        print(resource.getrusage(resource.RUSAGE_SELF))
        objgraph.show_most_common_types()
        after = hp.heap()
        leftover = after - before
        print(leftover)
        pool = concurrent.futures.ThreadPoolExecutor(max_workers=num_threads)

    count += 1

poc_memory_grows.py

import time

from cassandra.io.libevreactor import LibevConnection
import concurrent
import resource
from guppy import hpy
from cassandra.cluster import Cluster

connection_class=LibevConnection

import logging

log = logging.getLogger()
log.setLevel('INFO')
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s"))
log.addHandler(handler)


cluster = Cluster(("127.0.0.1",),  connection_class=connection_class)

session = cluster.connect()
session.execute('''
            CREATE KEYSPACE IF NOT EXISTS test3rf
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}''')

session.execute('''
            CREATE TABLE IF NOT EXISTS test3rf.test (
                k int PRIMARY KEY,
                v int )''')


def read():
    s = "SELECT * FROM test3rf.test"
    session.execute(s)

def write():
    s = "INSERT INTO test3rf.test (k, v) VALUES (1, 1)"
    session.execute(s)

num_threads = 50
copies = 2
count = 0


pool = concurrent.futures.ThreadPoolExecutor(max_workers=num_threads)
hp = hpy()
for i in range(200*25):

    for i in range(num_threads/2):
        pool.submit(read)
        pool.submit(write)
    time.sleep(0.0001)
    if not count % 200:
        print(resource.getrusage(resource.RUSAGE_SELF))

    count += 1

print("Waiting for pool to finish (this may take a while)")
pool.shutdown(wait=True)
print("\n\n\n\n======Finished the loop==========")

while 1:
    print(resource.getrusage(resource.RUSAGE_SELF))
    print(hp.heap())
    time.sleep(5)
    
print("The end")
print(gc.garbage)