memory_management_list.py

memory_management_list.py

import ctypes
import mmap
import os
import random
import string

import joblib
import numpy as np
import pandas as pd
import psutil
from flask import Flask


class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            instance = super().__call__(*args, **kwargs)
            cls._instances[cls] = instance
        return cls._instances[cls]


class SingletonJoblib(metaclass=SingletonMeta):
    def load_data(self):
        self.big_data = joblib.load('test.pkl', 'r')


class SingletonList(metaclass=SingletonMeta):
    def load_data(self):
        self.big_data = [item for item in range(10000000)]


class SingletonNumpy(metaclass=SingletonMeta):
    def load_data(self):
        self.big_data = np.array([[item, item] for item in range(10000000)])


class SingletonPandas(metaclass=SingletonMeta):
    def load_data(self):
        data = [['0', 1], ]
        for item in range(1000000):
            data.append([''.join([random.choice(string.ascii_lowercase) for a in range(3)]), item])
        df = pd.DataFrame(data, columns=['Name', 'Age'])
        self.big_data = df


class SingletonMmap(metaclass=SingletonMeta):
    def load_data(self):
        buf_len = 100000000
        buffer = mmap.mmap(-1, length=buf_len, access=mmap.ACCESS_WRITE)
        buffer[0:buf_len] = b"a" * buf_len
        self.big_data = buffer


def process_memory_usage(print_prefix):
    current_process = psutil.Process()
    memory = current_process.memory_info().rss
    print(f'{print_prefix} {int(memory / (1024 * 1024))} MB')


def system_memory_usage(print_prefix):
    print(f'{print_prefix} ({os.getpid()})->System used memory: {int(psutil.virtual_memory().used / (1024 * 1024))} MB')


def sharing_with_mmap():
    big_data = SingletonMmap().big_data
    system_memory_usage('')
    for i in range(len(big_data)):
        ref = big_data[i]

    system_memory_usage('')


def sharing_with_list():
    system_memory_usage('before ref')
    big_data = SingletonList().big_data
    print(ctypes.c_long.from_address(id(big_data)).value)
    ref1 = big_data[0]
    ref2 = big_data[0]
    print(ctypes.c_long.from_address(id(big_data)).value)
    system_memory_usage('ref only some elements')
    for item in big_data:
        pass
    system_memory_usage('after iteration')


def sharing_with_numpy():
    system_memory_usage('')
    big_data = SingletonNumpy().big_data
    print(ctypes.c_long.from_address(id(big_data)).value)
    ref1 = big_data[0]
    ref2 = big_data[0]
    print(big_data[0])
    print(ctypes.c_long.from_address(id(big_data)).value)
    system_memory_usage('')
    for item in big_data:
        pass
    system_memory_usage('')


def sharing_with_pandas():
    system_memory_usage('')
    big_data = SingletonPandas().big_data
    print(ctypes.c_long.from_address(id(big_data)).value)
    ref1 = big_data.iloc[0]
    ref2 = big_data.iloc[0]
    print(ctypes.c_long.from_address(id(big_data)).value)
    system_memory_usage('')
    for index, row in big_data.iterrows():
        pass
    system_memory_usage('')


def sharing_with_joblib():
    system_memory_usage('')
    big_data = SingletonJoblib().big_data
    print(ctypes.c_long.from_address(id(big_data)).value)
    ref1 = big_data[0]
    ref2 = big_data[0]
    print(ctypes.c_long.from_address(id(big_data)).value)
    system_memory_usage('')
    for x in big_data:
        pass
    system_memory_usage('')


system_memory_usage('Before load')
# Uncomment the the one to be used
# SingletonJoblib().load_data()
SingletonList().load_data()
# SingletonNumpy().load_data()
# SingletonPandas().load_data()
# SingletonMmap().load_data()
system_memory_usage('After load')

app = Flask(__name__)


@app.route('/joblib')
def joblib_view():
    sharing_with_joblib()
    return "Hello World!"


@app.route('/list')
def list_view():
    sharing_with_list()
    return "Hello World!"


@app.route('/numpy')
def numpy_view():
    sharing_with_numpy()
    return "Hello World!"


@app.route('/pandas')
def pandas_view():
    sharing_with_pandas()
    return "Hello World!"


@app.route('/mmap')
def mmap_view():
    sharing_with_mmap()
    return "Hello World!"