pandaserver/taskbuffer/DBProxyPool.py

0001 """
0002 pool for DBProxies
0003
0004 """
0005
0006 try:
0007     from Queue import Queue
0008 except ImportError:
0009     from queue import Queue
0010
0011 import os
0012 import random
0013 import time
0014 from contextlib import contextmanager
0015 from threading import Lock
0016
0017 from pandacommon.pandalogger.PandaLogger import PandaLogger
0018
0019 from pandaserver.config import panda_config
0020 from pandaserver.taskbuffer import OraDBProxy as DBProxy
0021 from pandaserver.taskbuffer.ConBridge import ConBridge
0022
0023 # logger
0024 _logger = PandaLogger().getLogger("DBProxyPool")
0025
0026
0027 class DBProxyPool:
0028     def __init__(self, dbhost, dbpasswd, nConnection, useTimeout=False, dbProxyClass=None):
0029         # crate lock for callers
0030         self.lock = Lock()
0031         self.callers = []
0032         # create Proxies
0033         _logger.debug("init")
0034         self.proxyList = Queue(nConnection)
0035         self.connList = []
0036         for i in range(nConnection):
0037             _logger.debug(f"connect -> {i} ")
0038             if dbProxyClass is not None:
0039                 proxy = dbProxyClass()
0040             elif useTimeout and hasattr(panda_config, "usedbtimeout") and panda_config.usedbtimeout is True:
0041                 """
0042                 ConBridge allows having database interactions in separate processes and killing them independently when interactions are stalled.
0043                 This avoids clogged httpd processes due to stalled database accesses.
0044                 """
0045                 proxy = ConBridge()
0046             else:
0047                 proxy = DBProxy.DBProxy()
0048                 self.connList.append(proxy)
0049             iTry = 0
0050             while True:
0051                 if proxy.connect(dbhost, dbpasswd, dbtimeout=60):
0052                     break
0053                 iTry += 1
0054                 _logger.debug(f"failed -> {i} : try {iTry}")
0055                 time.sleep(random.randint(60, 90))
0056             self.proxyList.put(proxy)
0057             time.sleep(1)
0058         # get PID
0059         self.pid = os.getpid()
0060         _logger.debug("ready")
0061
0062     # return a free proxy. this method blocks until a proxy is available
0063     def getProxy(self):
0064         # time how long it took to get a proxy
0065         start_time = time.time()
0066
0067         # get proxy
0068         proxy = self.proxyList.get()
0069         # wake up connection
0070         proxy.wakeUp()
0071
0072         end_time = time.time()
0073         elapsed_time = end_time - start_time
0074         _logger.debug(f"Getting proxy took: {elapsed_time} seconds")
0075
0076         return proxy
0077
0078     # put back a proxy
0079     def putProxy(self, proxy):
0080         self.proxyList.put(proxy)
0081
0082     # context manager for getting DBProxy
0083     @contextmanager
0084     def get(self):
0085         proxy = self.getProxy()
0086         try:
0087             yield proxy
0088         finally:
0089             self.putProxy(proxy)
0090
0091     # cleanup
0092     def cleanup(self):
0093         _logger.debug("cleanup start")
0094         [c.cleanup() for c in self.connList]
0095         _logger.debug("cleanup done")