Circuits memory leak

Circuits call

#!/usr/bin/env python3
import resource
import matplotlib.pyplot as plt

from circuits import Component, Debugger, Event


class dummy(Event):
    """Dummy event"""

class value(Event):
    """Graph value event"""
    def __init__(self, event_cnt, mem):
        super(value, self).__init__(event_cnt, mem)

class graph(Event):
    """Event that request graphing"""


class MemoryLeakTest(Component):
    nb_test = 1000

    def init(self, *args, **kwargs):
        self.start_mem = self.get_mem()
        self.values = [self.start_mem]


    def started(self, *args, **kwargs):
        for cnt in range(self.nb_test):
            yield self.call(dummy())
            self.values.append(self.get_mem())

        plt.plot(self.values)
        plt.savefig('call.png')

        print('Call count: %d' % self.nb_test)
        print('Memory used: %f' % (self.get_mem() - self.start_mem))

        raise SystemExit(0)

    def get_mem(self):
        return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

    def dummy(self):
        return 1


MemoryLeakTest().run()

Circuits fire

#!/usr/bin/env python3
import resource
import matplotlib.pyplot as plt

from circuits import Component, Debugger, Event


class dummy(Event):
    """Dummy event"""

class value(Event):
    """Graph value event"""
    def __init__(self, event_cnt, mem):
        super(value, self).__init__(event_cnt, mem)

class graph(Event):
    """Event that request graphing"""


class MemoryLeakTest(Component):
    nb_test = 1000

    def init(self, *args, **kwargs):
        self.start_mem = self.get_mem()
        self.values = [self.start_mem]


    def started(self, *args, **kwargs):
        for cnt in range(self.nb_test):
            self.fire(dummy())

        while len(self):
            self.values.append(self.get_mem())
            yield


        plt.plot(self.values)
        plt.savefig('fire.png')

        print('Call count: %d' % self.nb_test)
        print('Memory used: %f' % (self.get_mem() - self.start_mem))

        raise SystemExit(0)

    def get_mem(self):
        return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

    def dummy(self):
        return 1


MemoryLeakTest().run()