(self, cnt, src_db, dst_db)
Copy thread sync logic.
Copy thread sync logic.
Copy thread sync logic .
def sync_from_copy_thread(self, cnt, src_db, dst_db):
        "Copy thread sync logic."

        # somebody may have done remove-table in the meantime
        if self.copy_table_name not in self.table_map:
            self.log.error("copy_sync: lost table: %s", self.copy_table_name)
            return SYNC_EXIT

        # This operates on single table
        t = self.table_map[self.copy_table_name]

        if t.state == TABLE_DO_SYNC:
            # these settings may cause copy to miss right tick
            self.pgq_min_count = None
            self.pgq_min_interval = None

            # main thread is waiting, catch up, then handle over
            if self.cur_tick == t.sync_tick_id:
                self.change_table_state(dst_db, t, TABLE_OK)
                return SYNC_EXIT
            elif self.cur_tick < t.sync_tick_id:
                return SYNC_OK
            else:
                self.log.error("copy_sync: cur_tick=%d sync_tick=%d",
                                self.cur_tick, t.sync_tick_id)
                raise Exception('Invalid table state')
        elif t.state == TABLE_WANNA_SYNC:
            # wait for main thread to react
            return SYNC_LOOP
        elif t.state == TABLE_CATCHING_UP:

            # partition merging
            if t.copy_role in ('wait-replay', 'lead'):
                return SYNC_LOOP

            # copy just finished
            if t.dropped_ddl:
                self.restore_copy_ddl(t, dst_db)
                return SYNC_OK

            # is there more work?
            if self.work_state:
                return SYNC_OK

            # seems we have catched up
            self.change_table_state(dst_db, t, TABLE_WANNA_SYNC, self.cur_tick)
            return SYNC_LOOP
        elif t.state == TABLE_IN_COPY:
            # table is not copied yet, do it
            self.do_copy(t, src_db, dst_db)

            # forget previous value
            self.work_state = 1

            return SYNC_LOOP
        else:
            # nothing to do
            return SYNC_EXIT
def sync_from_copy_thread ( self , cnt , src_db , dst_db ) : # somebody may have done remove-table in the meantime if self . copy_table_name not in self . table_map : self . log . error ( "copy_sync: lost table: %s" , self . copy_table_name ) return SYNC_EXIT # This operates on single table t = self . table_map [ self . copy_table_name ] if t . state == TABLE_DO_SYNC : # these settings may cause copy to miss right tick self . pgq_min_count = None self . pgq_min_interval = None # main thread is waiting, catch up, then handle over if self . cur_tick == t . sync_tick_id : self . change_table_state ( dst_db , t , TABLE_OK ) return SYNC_EXIT elif self . cur_tick < t . sync_tick_id : return SYNC_OK else : self . log . error ( "copy_sync: cur_tick=%d sync_tick=%d" , self . cur_tick , t . sync_tick_id ) raise Exception ( 'Invalid table state' ) elif t . state == TABLE_WANNA_SYNC : # wait for main thread to react return SYNC_LOOP elif t . state == TABLE_CATCHING_UP : # partition merging if t . copy_role in ( 'wait-replay' , 'lead' ) : return SYNC_LOOP # copy just finished if t . dropped_ddl : self . restore_copy_ddl ( t , dst_db ) return SYNC_OK # is there more work? if self . work_state : return SYNC_OK # seems we have catched up self . change_table_state ( dst_db , t , TABLE_WANNA_SYNC , self . cur_tick ) return SYNC_LOOP elif t . state == TABLE_IN_COPY : # table is not copied yet, do it self . do_copy ( t , src_db , dst_db ) # forget previous value self . work_state = 1 return SYNC_LOOP else : # nothing to do return SYNC_EXIT