Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Example of `object pool' design pattern in Python
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Example of `object pool' design pattern
# Copyright (C) 2011 Radek Pazdera
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
class Resource:
""" Some resource, that clients need to use.
The resources usualy have a very complex and expensive
construction process, which is definitely not a case
of this Resource class in my example.
"""
__value = 0
def reset(self):
""" Put resource back into default setting. """
self.__value = 0
def setValue(self, number):
self.__value = number
def getValue(self):
return self.__value
class ObjectPool:
""" Resource manager.
Handles checking out and returning resources from clients.
It's a singleton class.
"""
__instance = None
__resources = list()
def __init__(self):
if ObjectPool.__instance != None:
raise NotImplemented("This is a singleton class.")
@staticmethod
def getInstance():
if ObjectPool.__instance == None:
ObjectPool.__instance = ObjectPool()
return ObjectPool.__instance
def getResource(self):
if len(self.__resources) > 0:
print "Using existing resource."
return self.__resources.pop(0)
else:
print "Creating new resource."
return Resource()
def returnResource(self, resource):
resource.reset()
self.__resources.append(resource)
def main():
pool = ObjectPool.getInstance()
# These will be created
one = pool.getResource()
two = pool.getResource()
one.setValue(10)
two.setValue(20)
print "%s = %d" % (one, one.getValue())
print "%s = %d" % (two, two.getValue())
pool.returnResource(one)
pool.returnResource(two)
one = None
two = None
# These resources will be reused
one = pool.getResource()
two = pool.getResource()
print "%s = %d" % (one, one.getValue())
print "%s = %d" % (two, two.getValue())
if __name__ == "__main__":
main()
@pabloruano

This comment has been minimized.

Copy link

pabloruano commented Nov 25, 2016

It would be nice to have something similar that allows a maximum number of instantiated Resources. If the limit is reached and getResource is called, the method would block until a call to returnResource() is made.

@hazelement

This comment has been minimized.

Copy link

hazelement commented Oct 12, 2018

You can use a queue instead of list which will do the blocking for ya.

@pavelsaman

This comment has been minimized.

Copy link

pavelsaman commented Dec 8, 2018

It would be nice to have something similar that allows a maximum number of instantiated Resources. If the limit is reached and getResource is called, the method would block until a call to returnResource() is made.

class Pool: 

    __instance = None
    __unused = list()
    __used = list()
    __max_pool_size = 3

    def __init__(self, max_pool_size=None):
        if Pool.__instance is not None:
            raise NotImplemented('Cannot create a new instance, this is a singleton class.')
        Pool.__instance = self
        if max_pool_size is not None:
            Pool.__max_pool_size = max_pool_size

    @staticmethod
    def get_pool_instance(): 
        if Pool.__instance is None:
            Pool.__instance = Pool()
        return Pool.__instance

    def create_new_instance(self, used=True):
        if len(self.__used) + len(self.__used) >= self.__max_pool_size:
            return None
        instance = Resource()
        if used:
            self.__used.append(instance)
        else:
            self.__unused.append(instance)
        return instance

    def get_unused(self):
        if len(self.__unused) > 0:
            i = self.__unused.pop(0)
            self.__used.append(i)
            return i
        else:
            return None

    def return_instance(self, instance):
        if instance in self.__used:
            if instance in self.__unused:
                return False
            self.__unused.append(instance)
            self.__used.remove(instance)
            return True
        else:
            return False

    def get_all_unused(self):
        return self.__unused.copy()

    def get_all_used(self):
        return self.__used.copy()

    def get_number_of_used(self):
        return len(self.__used)

    def get_number_of_unused(self):
        return len(self.__unused)

    def max_pool_size(self):
        return self.__max_pool_size
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.