Greenplum Python专用库gppylib学习——base.py

桃扇骨 2023-01-01 09:52 199阅读 0赞

base.py依赖的python包（Queue，threading，os，signal，subprocess/subprocess32，sys，time，warnings，paramiko，getpass），依赖的gp包（gplog，gpsubprocess，pygresql）。pygresql导入语句的是from pygresql.pg import DB，主要使用的DB是SQLCommand类，这个类先不用关注。gpsubprocess是对subprocess的封装，可以看到这里使用了两个子进程包gpsubprocess和subprocess。

# 代码分析 #

## WorkerPool类 ##

先看WorkPool类定义，类实例包含了存放worker实例的列表、存放带执行Command的work\_queue队列、存放执行完Command的completed\_queue队列。WorkerPool中的worker实例的数量是在构造函数就给定的，初始化后worker实例会一直运行，图中的start就是在构造函数中完成的。Worker实例从work\_queue队列中取工作项Command的函数是getNetWorkItem。

def getNextWorkItem(self):
        return self.work_queue.get(block=True)

Worker实例处理完成work\_queue中所有命令之后，取不到命令或取到的命令是halt\_command，或者任务池标志了should\_stop之后，使用markTaskDone函数告知WorkPool该任务完成。

def markTaskDone(self):
        self.work_queue.task_done()

clsSystemState.py的GpSystemStateProgram类中run函数中有base文件中类的使用示例，简化如下，通过这些示例来学习任务池的使用：

dispatchCount = 0
    pool = base.WorkerPool(parallelDegree) #parallelDegree给定worker个数
    for hostName, segments in ...:
      cmd = ...
      hostNameToCmd[hostName] = cmd
      pool.addCommand(cmd)
      dispatchCount+=1
    pool.wait_and_printdots(dispatchCount)
    hostNameToResults = { }
    for hostName, cmd in hostNameToCmd.iteritems():
      hostNameToResults[hostName] = cmd.decodeResults() #取出结果集
    pool.haltWork()

主要流程：通过addCommand函数向队列(work\_queue)中添加工作负载（WorkerPool类可以通过构造函数向队列中添加多个命令（列表形式），或者通过addCommand函数添加单个命令）。针对work\_queue队列由三种join的方法，代码如下。hostNameCmd是一个字典，键为hostName，值为cmd。通过cmd.decodeResults函数取出结果集。

def join(self):
        self.work_queue.join()
        return True
    def _join_work_queue_with_timeout(self, timeout):
            """ Queue.join() unfortunately doesn't take a timeout (see https://bugs.python.org/issue9634). Fake it here, with a solution inspired by notes on that bug report. XXX This solution uses undocumented Queue internals (though they are not underscore-prefixed...). """
        done_condition = self.work_queue.all_tasks_done
        done_condition.acquire()
        try:
          while self.work_queue.unfinished_tasks:
            if (timeout <= 0):
              # Timed out.
              return
              
            start_time = time.time()
            done_condition.wait(timeout)
            timeout -= (time.time() - start_time)
        finally:
            done_condition.release()
    def wait_and_printdots(self, command_count, quiet=True):
      while self.completed_queue.qsize() < command_count:
        time.sleep(1)
        
        if not quiet:
          sys.stdout.write(".")
          sys.stdout.flush()
      if not quiet:
        print " "
      self.join()

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]  
Queue/queue模块的类

属性 描述  
Queue(maxsize=0) 创建一个先入先出队列。如果给定最大值，则在队列没有空间时阻塞；否则，为无限队列  
LifoQueue(maxsize=0) 创建一个后入先出队列。如果给定最大值，则在队列没有空间时阻塞；否则，为无限队列  
PriorityQueue(maxsize=0) 创建一个优先级队列。如果给定最大值，则在队列没有空间时阻塞，否则，为无限队列  
Queue/queue异常  
属性 描述  
Empty  
当对孔队列调用get\*()方法时抛出异常  
Full 当对已满的队列调用put\*()方法时抛出异常

## Worker类 ##

worker类继承自threading模块中的Thread类，run函数先使用getNextWorkItem函数取得command，总共有四种情况：任务池中没有command，该Worker示例需要向任务池标记任务完成；如果取得的命令是pool.halt\_command，该Worker示例需要向任务池标记任务完成；如果任务池标记了should\_stop，该Worker示例需要向任务池标记任务完成；下面是正常流程，执行命令，并将命令放入任务池完成队列。

class Woker(Thread):
        ...
        def run(self):
            while True:
                try:
                    try:
                        self.cmd = self.pool.getNextWorkItem()
                    except TypeError:
                        # misleading exception raised during interpreter shutdown
                        return
                    # we must have got a command to run here
                    if self.cmd is None:
                        self.logger.debug("[%s] got a None cmd" % self.name)
                        self.pool.markTaskDone()
                    elif self.cmd is self.pool.halt_command:
                        self.logger.debug("[%s] got a halt cmd" % self.name)
                        self.pool.markTaskDone()
                        self.cmd = None
                        return
                    elif self.pool.should_stop:
                        self.logger.debug("[%s] got cmd and pool is stopped: %s" % (self.name, self.cmd))
                        self.pool.markTaskDone()
                        self.cmd = None
                    else:
                        self.logger.debug("[%s] got cmd: %s" % (self.name, self.cmd.cmdStr))
                        self.cmd.run()
                        self.logger.debug("[%s] finished cmd: %s" % (self.name, self.cmd))
                        self.pool.addFinishedWorkItem(self.cmd)
                        self.cmd = None
                except Exception, e:
                    self.logger.exception(e)
                    if self.cmd:
                        self.logger.debug("[%s] finished cmd with exception: %s" % (self.name, self.cmd))
                        self.pool.addFinishedWorkItem(self.cmd)
                        self.cmd = None
        def haltWork(self):
            self.logger.debug("[%s] haltWork" % self.name)  
            c = self.cmd
            if c is not None and isinstance(c, Command):
                c.interrupt()
                c.cancel()

threading模块的Thread类实例化表示一个执行线程的对象，拥有的数据属性name(线程名)、ident(线程的标识符)、daemon(布尔标志，表示这个线程是否是守护线程)，方法如下：  
Thread对象方法描述  
`_ini_(group=None, target=None, name=None, args=(), kwargs={}, verbose=None, daemon=None)` 实例化一个线程对象，需要有一个可调用的target，以及其参数args或kwargs。还可以传递name或group参数，不过后者还未实现。此外，verbose标志也是可以接受的。而daemon的值将会设定thread.daemon属性/标志

<table> 
 <thead> 
  <tr> 
   <th>成员</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>start()</td> 
   <td>开始执行该线程</td> 
  </tr> 
  <tr> 
   <td>run()</td> 
   <td>定义线程功能的方法（通常在子类中被应用开发者重写）</td> 
  </tr> 
  <tr> 
   <td>join(timeout=None)</td> 
   <td>直至启动的线程终止之前一直挂起；除非给出timeout(秒)，否则会一直阻塞</td> 
  </tr> 
  <tr> 
   <td>getName()</td> 
   <td>返回线程名</td> 
  </tr> 
  <tr> 
   <td>setName(name)</td> 
   <td>设定线程名</td> 
  </tr> 
  <tr> 
   <td>isAlivel/is_alive()</td> 
   <td>布尔标志，表示这个线程是否还存活</td> 
  </tr> 
  <tr> 
   <td>isDaemon()</td> 
   <td>如果是守护线程，则返回True；否则，返回False</td> 
  </tr> 
  <tr> 
   <td>setDaemon(daemonic)</td> 
   <td>把线程的守护标志设定为布尔值daemonic（必须在线程start()之前调用）</td> 
  </tr> 
 </tbody> 
</table>

使用Thread类可以有很多方法创建线程，这里介绍三种方法：1. 创建Thread的实例，传给它一个函数；2. 创建Thread的实例，传给它一个可调用的类实例；3.派生Thread的子类，并创建子类的实例。

## Command类 ##

Command类有两个执行函数runNoWait和run函数，runNoWait函数通过调用exec\_context.execute(self,wait=False)函数执行命令，并返回proc；run函数直接调用exec\_context.execute(self)函数。

def runNoWait(self):
            faultPoint = os.getenv('GP_COMMAND_FAULT_POINT')
            if not faultPoint or (self.name and not self.name.startswith(faultPoint)):
                self.exec_context.execute(self, wait=False)
                return self.exec_context.proc
        def run(self, validateAfter=False):
            faultPoint = os.getenv('GP_COMMAND_FAULT_POINT')
            if not faultPoint or (self.name and not self.name.startswith(faultPoint)):
                self.exec_context.execute(self)
            else:
                # simulate error
                self.results = CommandResult(1, 'Fault Injection', 'Fault Injection', False, True)
            if validateAfter:
                self.validate()
            pass

## ExecutionContext类 ##

以RemoteExecutionContext执行上下文类为例，参数是以以下方法处理的  
`keys = sorted(cmd.propagate_env_map.keys(), reverse=True) for k in keys: cmd.cmdStr = "%s=%s && %s" % (k, cmd.propagate_env_map[k], cmd.cmdStr)`，将参数序列化到cmdStr中。对于LocalExecutionContext来说，调用如下命令执行命令：`self.proc = gpsubprocess.Popen(cmd.cmdStr, env=None, shell=True,executable='/bin/bash',stdin=subprocess.PIPE,stderr=subprocess.PIPE,stdout=subprocess.PIPE, close_fds=True)`，对GP封装的subprocess模式请参见相应其他系列的博客。如果需要等待子进程，则调用`(rc, stdout_value, stderr_value) = self.proc.communicate2(input=self.stdin)`，然后使用`cmd.set_results(CommandResult( rc, "".join(stdout_value), "".join(stderr_value), self.completed, self.halt)) def cancel(self, cmd):`封装命令返回的结果。

class LocalExecutionContext(ExecutionContext):
        proc = None
        halt = False
        completed = False
        def __init__(self, stdin):
            ExecutionContext.__init__(self)
            self.stdin = stdin
            pass
        def execute(self, cmd, wait=True):
            # prepend env. variables from ExcecutionContext.propagate_env_map
            # e.g. Given {'FOO': 1, 'BAR': 2}, we'll produce "FOO=1 BAR=2 ..."
            # also propagate env from command instance specific map
            keys = sorted(cmd.propagate_env_map.keys(), reverse=True)
            for k in keys:
                cmd.cmdStr = "%s=%s && %s" % (k, cmd.propagate_env_map[k], cmd.cmdStr)
            # executable='/bin/bash' is to ensure the shell is bash. bash isn't the
            # actual command executed, but the shell that command string runs under.
            self.proc = gpsubprocess.Popen(cmd.cmdStr, env=None, shell=True,executable='/bin/bash',stdin=subprocess.PIPE,stderr=subprocess.PIPE,stdout=subprocess.PIPE, close_fds=True)
            cmd.pid = self.proc.pid
            if wait:
                (rc, stdout_value, stderr_value) = self.proc.communicate2(input=self.stdin)
                self.completed = True
                cmd.set_results(CommandResult(
                    rc, "".join(stdout_value), "".join(stderr_value), self.completed, self.halt))
        def cancel(self, cmd):
            if self.proc:
                try:
                    os.kill(self.proc.pid, signal.SIGTERM)
                except OSError:
                    pass
        def interrupt(self, cmd):
            self.halt = True
            if self.proc:
                self.proc.cancel()
    
    class RemoteExecutionContext(LocalExecutionContext):
        trail = set()
        def __init__(self, targetHost, stdin, gphome=None):
            LocalExecutionContext.__init__(self, stdin)
            self.targetHost = targetHost
            if gphome:
                self.gphome = gphome
            else:
                self.gphome = GPHOME
        def execute(self, cmd):
            # prepend env. variables from ExcecutionContext.propagate_env_map
            # e.g. Given {'FOO': 1, 'BAR': 2}, we'll produce "FOO=1 BAR=2 ..."
            self.__class__.trail.add(self.targetHost)
            # also propagate env from command instance specific map
            keys = sorted(cmd.propagate_env_map.keys(), reverse=True)
            for k in keys:
                cmd.cmdStr = "%s=%s && %s" % (k, cmd.propagate_env_map[k], cmd.cmdStr)
            # Escape " for remote execution otherwise it interferes with ssh
            cmd.cmdStr = cmd.cmdStr.replace('"', '\\"')
            cmd.cmdStr = "ssh -o StrictHostKeyChecking=no -o ServerAliveInterval=60 " \
                         "{targethost} \"{gphome} {cmdstr}\"".format(targethost=self.targetHost,
                                                                     gphome=". %s/greenplum_path.sh;" % self.gphome,
                                                                     cmdstr=cmd.cmdStr)
            LocalExecutionContext.execute(self, cmd)
            if (cmd.get_results().stderr.startswith('ssh_exchange_identification: Connection closed by remote host')):
                self.__retry(cmd)
            pass
        def __retry(self, cmd, count=0):
            if count == SSH_MAX_RETRY:
                return
            time.sleep(SSH_RETRY_DELAY)
            LocalExecutionContext.execute(self, cmd)
            if (cmd.get_results().stderr.startswith('ssh_exchange_identification: Connection closed by remote host')):
                self.__retry(cmd, count + 1)

而RemoteExecutionContext先对参数进行格式化后，还需要对双引号使用双斜线进行替换后，然后添加ssh相关命令选项，代码如下所示：`cmd.cmdStr = cmd.cmdStr.replace('"', '\\"') cmd.cmdStr = "ssh -o StrictHostKeyChecking=no -o ServerAliveInterval=60 {targethost} \"{gphome} {cmdstr}\"".format(targethost=self.targetHost, gphome=". %s/greenplum_path.sh;" % self.gphome, cmdstr=cmd.cmdStr)`，然后调用LocalExecutionContext父类的execute函数。如果返回的结果包含ssh\_exchange\_identification: Connection closed by remote host，则需要进行等待相应的时间，然后进行重试。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]

## Controller-Worker架构模式 ##

![图1 架构模式][1]  
辅助说明：  
Controller-Worker是一种组合架构模式，Controller基于Client的参数动态生成Woker数量，并控制Woker的生命周期，如创建和终止。  
Controller属性：  
Controller事先知道自身拥有的Woker类型。  
Controller依赖一个工作任务池，通过工作任务池Controller监控整体任务执行情况。  
Worker属性：  
Worker并行消费工作任务池中任务，并把执行结果返回到任务池中。  
Worker彼此间没有任何耦合。  
![图4 架构数据流][4]  
辅组说明：  
Controller通过WorkerPool和Worker进行命令传递。  
Controller通过超时机制，保证最后一定有命令结果返回给Client  
Controller通过halt命令，停止所有的Woker  
Worker采用Thread方式来实现。  
Worker1、Worker2、WorkerN无差别，根据获取的Cmd，通过ssh方式在对应的Host执行命令。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]: /images/20221120/be7ad04ed75045a6a6c86c7bb9c3d0c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]: /images/20221120/0cd41ba533a8443ebfc1aefba75a541e.png
[1]: https://img-blog.csdnimg.cn/20210101222425281.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg==,size_16,color_FFFFFF,t_70
[4]: https://img-blog.csdnimg.cn/20210101222520637.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg==,size_16,color_FFFFFF,t_70