(self, cnt, src_db, dst_db)
return ret
Main thread sync logic.
Main thread sync logic.
Main thread sync logic .
def sync_from_main_thread(self, cnt, src_db, dst_db):
        "Main thread sync logic."

        # This operates on all table, any amount can be in any state

        ret = SYNC_OK

        if cnt.do_sync:
            # wait for copy thread to catch up
            ret = SYNC_LOOP

        # we need to do wanna-sync->do_sync with small batches
        need_dsync = False
        dsync_ok = True
        if self.pgq_min_interval or self.pgq_min_count:
            dsync_ok = False
        elif self.dsync_backup and self.dsync_backup[0] >= self.cur_tick:
            dsync_ok = False

        # now check if do-sync is needed
        for t in self.get_tables_in_state(TABLE_WANNA_SYNC):
            # copy thread wants sync, if not behind, do it
            if self.cur_tick >= t.sync_tick_id:
                if dsync_ok:
                    self.change_table_state(dst_db, t, TABLE_DO_SYNC, self.cur_tick)
                    ret = SYNC_LOOP
                else:
                    need_dsync = True

        # tune batch size if needed
        if need_dsync:
            if self.pgq_min_count or self.pgq_min_interval:
                bak = (self.cur_tick, self.pgq_min_count, self.pgq_min_interval)
                self.dsync_backup = bak
                self.pgq_min_count = None
                self.pgq_min_interval = None
        elif self.dsync_backup:
            self.pgq_min_count = self.dsync_backup[1]
            self.pgq_min_interval = self.dsync_backup[2]
            self.dsync_backup = None

        # now handle new copies
        npossible = self.parallel_copies - cnt.get_copy_count()
        if cnt.missing and npossible > 0:
            pmap = self.get_state_map(src_db.cursor())
            src_db.commit()
            for t in self.get_tables_in_state(TABLE_MISSING):
                if 'copy_node' in t.table_attrs:
                    # should we go and check this node?
                    pass
                else:
                    # regular provider is used
                    if t.name not in pmap:
                        self.log.warning("Table %s not available on provider", t.name)
                        continue
                    pt = pmap[t.name]
                    if pt.state != TABLE_OK: # or pt.custom_snapshot: # FIXME: does snapsnot matter?
                        self.log.info("Table %s not OK on provider, waiting", t.name)
                        continue

                # don't allow more copies than configured
                if npossible == 0:
                    break
                npossible -= 1

                # drop all foreign keys to and from this table
                self.drop_fkeys(dst_db, t.dest_table)

                # change state after fkeys are dropped thus allowing
                # failure inbetween
                self.change_table_state(dst_db, t, TABLE_IN_COPY)

                # the copy _may_ happen immediately
                self.launch_copy(t)

                # there cannot be interesting events in current batch
                # but maybe there's several tables, lets do them in one go
                ret = SYNC_LOOP

        return ret
def sync_from_main_thread ( self , cnt , src_db , dst_db ) : # This operates on all table, any amount can be in any state ret = SYNC_OK if cnt . do_sync : # wait for copy thread to catch up ret = SYNC_LOOP # we need to do wanna-sync->do_sync with small batches need_dsync = False dsync_ok = True if self . pgq_min_interval or self . pgq_min_count : dsync_ok = False elif self . dsync_backup and self . dsync_backup [ 0 ] >= self . cur_tick : dsync_ok = False # now check if do-sync is needed for t in self . get_tables_in_state ( TABLE_WANNA_SYNC ) : # copy thread wants sync, if not behind, do it if self . cur_tick >= t . sync_tick_id : if dsync_ok : self . change_table_state ( dst_db , t , TABLE_DO_SYNC , self . cur_tick ) ret = SYNC_LOOP else : need_dsync = True # tune batch size if needed if need_dsync : if self . pgq_min_count or self . pgq_min_interval : bak = ( self . cur_tick , self . pgq_min_count , self . pgq_min_interval ) self . dsync_backup = bak self . pgq_min_count = None self . pgq_min_interval = None elif self . dsync_backup : self . pgq_min_count = self . dsync_backup [ 1 ] self . pgq_min_interval = self . dsync_backup [ 2 ] self . dsync_backup = None # now handle new copies npossible = self . parallel_copies - cnt . get_copy_count ( ) if cnt . missing and npossible > 0 : pmap = self . get_state_map ( src_db . cursor ( ) ) src_db . commit ( ) for t in self . get_tables_in_state ( TABLE_MISSING ) : if 'copy_node' in t . table_attrs : # should we go and check this node? pass else : # regular provider is used if t . name not in pmap : self . log . warning ( "Table %s not available on provider" , t . name ) continue pt = pmap [ t . name ] if pt . state != TABLE_OK : # or pt.custom_snapshot: # FIXME: does snapsnot matter? self . log . info ( "Table %s not OK on provider, waiting" , t . name ) continue # don't allow more copies than configured if npossible == 0 : break npossible -= 1 # drop all foreign keys to and from this table self . drop_fkeys ( dst_db , t . dest_table ) # change state after fkeys are dropped thus allowing # failure inbetween self . change_table_state ( dst_db , t , TABLE_IN_COPY ) # the copy _may_ happen immediately self . launch_copy ( t ) # there cannot be interesting events in current batch # but maybe there's several tables, lets do them in one go ret = SYNC_LOOP return ret