GBase8a Rsynctool 同步工具网络连接超时时间修改方法

1.背景说明：

主从集群间通过Rsynctool进行数据同步时，因主从集群间网络带宽较低且单表同步数据量较大时，同步工具会打屏或向日志中输出多次retry信息，默认情况下输出60次retry后将终止该表的同步，造成该表同步失败。
打屏输出超时日志类似如下：
python api recv timeout,retry [1/60]
python api recv timeout,retry [2/60]
python api recv timeout,retry [3/60]
...
超时原因是Rsynctool工具Python代码复用了GBase8a Python API中的网络连接部分代码，其中对网络连接的超时进行了初始化设置，并在调用更底层的python语言的socket对象代码时，使用了socket.recv()函数读取socket连接缓冲区中的内容，recv方法主要用于从一个已连接的套接字接收数据，它会阻塞当前线程，直到有数据到达或者超时产生socket exception。GBase8a Python API中对这种socket.revc()因达到超时设置而产生exception的情况进行了retry的处理，最多进行60次retry，超过retry的最大次数则会退出socket连接，从而导致表同步失败。
本文从Rsynctool python代码分析的角度解释超时设置原理，并给出两个方案设置更长的超时时间，防止大表同步时因网络带宽较低导致达到连接超时报错的问题。

2.代码分析：

GBase8a Rsynctool工具在网络层复用了GBase8a Python API连接8a Server部分的代码，这部分Python API代码主要在工具 gcluster_rsynctool/GBaseConnector/ 目录中
其中：
gcluster_rsynctool/GBaseConnector/GBaseSocket.py 中定义了GBaseSocket类，其中继承了python socket类中的一些方法，也包括了初始化socket类中成员timeout的方法：

class GBaseSocket(object):
   '''
       GBase socket communicating with GBase Server
   '''
   def __init__(self, host='127.0.0.1', port=5258, timeout=None):
       '''
       Constructor
       '''
       self._host = host
       self._port = port
       self._timeout = timeout
       self._gbase_socket = None
       self._packet_number = -1

       self.open_tcp()
...
   def open_tcp(self):
       '''
          Function    : open_tcp
          Arguments   :
          Return      :
          Description : create a tcp socket
       '''
       GBASELOGGER.debug("Enter: open_tcp.")
       addrinfo = None
       try:
           addrinfos = socket.getaddrinfo(self._host, self._port, 0, socket.SOCK_STREAM)
           for info in addrinfos:
               if info[0] == socket.AF_INET6:
                   addrinfo = info
                   break
               elif info[0] == socket.AF_INET:
                   addrinfo = info
                   break
           (family, socktype, proto, canonname, sockaddr) = addrinfo
       except socket.gaierror, err:
           GBASELOGGER.debug("Exit: open_tcp.")
           raise GBaseError.InterfaceError(errno=2003,  values=(self._get_address(), err[1]))

       try:
           self._gbase_socket = socket.socket(family, socktype, proto)
           self._gbase_socket.settimeout(self._timeout)
           self._gbase_socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
           self._gbase_socket.setsockopt(socket.SOL_SOCKET,socket.SO_RCVTIMEO,struct.pack('LL',self._timeout,0))
           self._gbase_socket.setsockopt(socket.SOL_SOCKET,socket.SO_SNDTIMEO,struct.pack('LL',self._timeout,0))
           self._gbase_socket.connect(sockaddr)
       except socket.gaierror, err:
           GBASELOGGER.debug("Exit: open_tcp.")
           raise GBaseError.InterfaceError(errno=2003, values=(self._get_address(), err[1]))
       except socket.error, err:
           try:
               msg = err.errno
               if msg is None:
                   msg = str(err)
           except AttributeError:
               msg = str(err)
           GBASELOGGER.debug("Exit: open_tcp.")
           raise GBaseError.InterfaceError(
               errno=2003, values=(self._get_address(), msg))
       except StandardError, err:
           GBASELOGGER.debug("Exit: open_tcp.")
           raise GBaseError.InterfaceError(errmsg = '%s' % err)
       except:
           GBASELOGGER.debug("Exit: open_tcp.")
           raise
       GBASELOGGER.debug("Exit: open_tcp.")

在gcluster_rsynctool/GBaseConnector/GBaseSocket.py中的recvdata()成员方法中设计了Rsynctool接收从GBase8a Server返回的消息时读取缓冲区buf中数据的逻辑：
其中_gbase_socket.recv 方法主要用于从一个已连接的套接字接收数据，它会阻塞当前线程，直到有数据到达或者超时，当出现socket.timeout时根据是否达到MAX_RETRY_TIMES决定是否退出接收数据并报异常，当未达到MAX_RETRY_TIMES时将打屏输出python api recv timeout,retry [当时retry次数/60]，当达到MAX_RETRY_TIMES后将报错退出。

 def recvdata(self,len):
       retry_times=1
       buf = ''
       while True:
           try:
               buf = self._gbase_socket.recv(len)
               break
           except (socket.timeout,Exception), err:
               if retry_times > MAX_RETRY_TIMES:
                   #raise err
                   break
               print "python api recv timeout,retry [%d/%d]"%(retry_times,MAX_RETRY_TIMES)
               retry_times = retry_times + 1
       return buf

在gcluster_rsynctool/GBaseConnector/GBaseConnection.py 文件中，通过定义GBaseConnection类来封装连接GBase Server部分的逻辑，GBaseConnection类中的成员定义为GBaseSocket类的实例，因此GBaseConnection与GBaseSocket有继承关系。GBaseConnection类的代码对GBaseSocket实例成员的初始化部分设定了默认的sockettimeout值，为30s。

class GBaseConnection(object):
   '''
   ************************************************************
       Module      : GBaseConnection
       Function    : Create connection to GBase server
       Corporation : General Data Technology CO,. Ltd.
       Team        : DMD, Interface Team
       Author      : wq
       Date        : 2013-7
       Version     : v1.0
       Description : Create connection to GBase server. And execute
                     SQL and fetch rows.
   ************************************************************
   '''
   def __init__(self, **kwargs):
       '''
       ************************************************************
           Function    : __init__
           Arguments   : 1) **kwargs (dict) : connection arguments
                                               { 'user': gbase,
                                                 'port': 5258  }
           Return      :
           Description : GBaseConnection construct function
       ************************************************************
       '''
       self._host = '127.0.0.1'
       self._user = ''
       self._password = ''
       self._database = ''
       self._port = 5258
       self.charset = None
       self._use_unicode = True
       self._collation = None
       self._connection_timeout = 30  #socket连接默认值 单位s
       self._autocommit = True
       self._time_zone = None
       self._sql_mode = None
       self._get_warnings = False
       self._raise_on_warnings = False

       self._client_flags = ClientFlag.get_default()
       self._charset_id = 33  # utf8

       self._socket = None
       self._protocol = None
       self._hello_res = None
       self._server_version = None
       self._have_next_result = False
       self._unread_result = False

       # Initial log system
       GBaseLogger.InitLog(**kwargs)

       if len(kwargs) > 0:
           self.connect(**kwargs)
...
   def _get_connection(self):
       '''
       ************************************************************
           Function    : _get_connection
           Arguments   :
           Return      :
           Description : Only support TCP socket
       ************************************************************
       '''
       GBASELOGGER.debug('Enter: _get_connection')
       socket = None
       hosts = self._host
       errmsg = ''
       errno = 0
       if isinstance(self._host, str):
           hosts = [self._host]
       for host in hosts:
           try:
               socket = GBaseSocket(host=host, port=self._port,
                                    timeout=self.connection_timeout)  #初始化GBaseSocket实例，用30s默认值初始化socket.timeout值
           except GBaseError.InterfaceError, err:
               errmsg = str(err)
               errno = err.errno
               GBASELOGGER.error(str(err))
       if socket is None:
           raise GBaseError.InterfaceError(errmsg, errno)
       GBASELOGGER.debug('Exit: _get_connection')
       return socket

在工具上层代码gcluster_rsynctool/GCConnection.py中，定义了CSqlClient()类来封装Rsycntool工具连接GBase8a Server并执行sql语句的逻辑，其代码引用了GBaseConnection类，因此也集成了其Socket.timeout的默认值30s。

class CSqlClient(object):
   def __init__(self,host,port,user,passwd,charset='utf8',timeout=60):
       self.conn=None
       try:
           #config = {'user': user, 'password': passwd, 'host': host, 'port': port, 'charset': 'utf8','connection_timeout':timeout}
           config = {'user': user, 'password': passwd, 'host': host, 'port': port, 'charset': charset,'connection_timeout':timeout}
           #self.conn = connect(host=host, user=user, passwd=passwd, port=port, connection_timeout=timeout)
           self.conn = GBaseConnection(**config)
       except GBaseError.InterfaceError,e:
           raise GCException(str(e))
       except Exception,e:
           raise GCException(str(e))
       self.cursor = self.conn.cursor()

   def __del__(self):
       if self.conn != None:
           self.conn.close()

   def ExecSql(self,sql):
       GBaseError
       try:
           self.cursor.execute(sql)
       except GBaseError.InterfaceError,e:
           raise GCException(str(e))
       except GBaseError.DatabaseError,e:
           raise GCException(str(e))
       except GBaseError.InternalError,e:
           raise GCException(str(e))
       except Exception,e:
           raise GCException(str(e))

   def GetResultRows(self):
       return self.cursor.fetchall()

   def close(self):
       #self.cursor.close()
       self.conn.close()

综上，当初始化socket超时参数timeout值设定为默认值30s时，如果一张表的同步时间超过30s * MAX_RETRY_TIMES（默认值为60）= 1800s时，通过Rsynctool工具代码gcluster_rsynctool/GCConnection.py中的CSqlClient.ExecSql下发的同步任务指令将因超时而退出，从而导致同步失败。
如果希望调整代码，达到增大表同步超时时间的目的，可以修改socket超时时间的默认值30s，在gcluster_rsynctool/GBaseConnector/GBaseConnection.py中的如下位置：

或者修改gcluster_rsynctool/GBaseConnector/GBaseRetryTimes.py，将MAX_RETRY_TIMES=60改大。
最终大表同步数据的超时时间为：self._connection_timeout * MAX_RETRY_TIMES的值。