(self, src_db, dst_db)
Send local watermark to provider.
Send local watermark to provider.
Send local watermark to provider .
def publish_local_wm(self, src_db, dst_db):
        """Send local watermark to provider.
        """

        t = time.time()
        if t - self.local_wm_publish_time < self.local_wm_publish_period:
            return

        st = self._worker_state
        wm = st.local_watermark
        if st.sync_watermark:
            # dont send local watermark upstream
            wm = self.batch_info['prev_tick_id']
        elif wm > self.batch_info['cur_tick_id']:
            # in wait-behind-leaf case, the wm from target can be
            # ahead from source queue, use current batch then
            wm = self.batch_info['cur_tick_id']

        self.log.debug("Publishing local watermark: %d", wm)
        src_curs = src_db.cursor()
        q = "select * from pgq_node.set_subscriber_watermark(%s, %s, %s)"
        src_curs.execute(q, [self.pgq_queue_name, st.node_name, wm])
        src_db.commit()

        # if next part fails, dont repeat it immediately
        self.local_wm_publish_time = t

        if st.sync_watermark and self.real_global_wm is not None:
            # instead sync 'global-watermark' with specific nodes
            dst_curs = dst_db.cursor()
            nmap = self._get_node_map(dst_curs)
            dst_db.commit()

            # local lowest
            wm = st.local_watermark

            # the global-watermark in subtree can stay behind
            # upstream global-watermark, but must not go ahead
            if self.real_global_wm < wm:
                wm = self.real_global_wm

            for node in st.wm_sync_nodes:
                if node == st.node_name:
                    continue
                if node not in nmap:
                    # dont ignore missing nodes - cluster may be partially set up
                    self.log.warning('Unknown node in sync_watermark list: %s', node)
                    return
                n = nmap[node]
                if n['dead']:
                    # ignore dead nodes
                    continue
                wmdb = self.get_database('wmdb', connstr = n['node_location'], autocommit = 1, profile = 'remote')
                wmcurs = wmdb.cursor()
                q = 'select local_watermark from pgq_node.get_node_info(%s)'
                wmcurs.execute(q, [self.queue_name])
                row = wmcurs.fetchone()
                if not row:
                    # partially set up node?
                    self.log.warning('Node not working: %s', node)
                elif row['local_watermark'] < wm:
                    # keep lowest wm
                    wm = row['local_watermark']
                self.close_database('wmdb')

            # now we have lowest wm, store it
            q = "select pgq_node.set_global_watermark(%s, %s)"
            dst_curs.execute(q, [self.queue_name, wm])
            dst_db.commit()
def publish_local_wm ( self , src_db , dst_db ) : t = time . time ( ) if t - self . local_wm_publish_time < self . local_wm_publish_period : return st = self . _worker_state wm = st . local_watermark if st . sync_watermark : # dont send local watermark upstream wm = self . batch_info [ 'prev_tick_id' ] elif wm > self . batch_info [ 'cur_tick_id' ] : # in wait-behind-leaf case, the wm from target can be # ahead from source queue, use current batch then wm = self . batch_info [ 'cur_tick_id' ] self . log . debug ( "Publishing local watermark: %d" , wm ) src_curs = src_db . cursor ( ) q = "select * from pgq_node.set_subscriber_watermark(%s, %s, %s)" src_curs . execute ( q , [ self . pgq_queue_name , st . node_name , wm ] ) src_db . commit ( ) # if next part fails, dont repeat it immediately self . local_wm_publish_time = t if st . sync_watermark and self . real_global_wm is not None : # instead sync 'global-watermark' with specific nodes dst_curs = dst_db . cursor ( ) nmap = self . _get_node_map ( dst_curs ) dst_db . commit ( ) # local lowest wm = st . local_watermark # the global-watermark in subtree can stay behind # upstream global-watermark, but must not go ahead if self . real_global_wm < wm : wm = self . real_global_wm for node in st . wm_sync_nodes : if node == st . node_name : continue if node not in nmap : # dont ignore missing nodes - cluster may be partially set up self . log . warning ( 'Unknown node in sync_watermark list: %s' , node ) return n = nmap [ node ] if n [ 'dead' ] : # ignore dead nodes continue wmdb = self . get_database ( 'wmdb' , connstr = n [ 'node_location' ] , autocommit = 1 , profile = 'remote' ) wmcurs = wmdb . cursor ( ) q = 'select local_watermark from pgq_node.get_node_info(%s)' wmcurs . execute ( q , [ self . queue_name ] ) row = wmcurs . fetchone ( ) if not row : # partially set up node? self . log . warning ( 'Node not working: %s' , node ) elif row [ 'local_watermark' ] < wm : # keep lowest wm wm = row [ 'local_watermark' ] self . close_database ( 'wmdb' ) # now we have lowest wm, store it q = "select pgq_node.set_global_watermark(%s, %s)" dst_curs . execute ( q , [ self . queue_name , wm ] ) dst_db . commit ( )