Python gevent.pool 模块，join() 实例源码

def run(self):
        lookups = self.rectypes or ['CNAME', 'A', 'AAAA']
        dnsname = self.domain
        if self.name is None:
            # Top-level, needs extra queries
            lookups += ['MX', 'SOA', 'NS', 'SRV', 'TXT', 'SPF', 'RRSIG', 'DS',
                        'DLV', 'DNSKEY']
        else:
            dnsname = '.'.join([self.name, dnsname])
        for query_type in set(lookups):
            resp = None
            LOG.debug("Checking %s %s", dnsname, query_type)
            try:
                resp = self.bruter.query(dnsname, query_type)
            except DNSException:
                continue
            except Exception:
                LOG.exception("While resolving %s %s", dnsname, query_type)
                continue
            self.bruter.on_result(self.domain, self.name, query_type, resp)
        self.bruter.on_finish()

def record_testreport(self, testreport):
        assert not self.testcase
        names = mangle_test_address(testreport.nodeid)
        classnames = names[:-1]
        if self.xml.prefix:
            classnames.insert(0, self.xml.prefix)
        attrs = {
            "classname": ".".join(classnames),
            "name": bin_xml_escape(names[-1]),
            "file": testreport.location[0],
        }
        if testreport.location[1] is not None:
            attrs["line"] = testreport.location[1]
        if hasattr(testreport, "url"):
            attrs["url"] = testreport.url
        self.attrs = attrs

def run(self):
        if self.beanstalk:
            generator = self.beanstalk.get_workgenerator(self)
        else:
            generator = ListWorkGenerator(self)

        pool = gevent.pool.Pool(self.options.concurrency)
        self.finished = 0
        if self.progress:
            self.progress.start(generator.total)

        try:
            for worker in generator.getall():
                pool.add(gevent.spawn(worker.run))
        except KeyboardInterrupt:
            print("Ctrl+C caught... stopping")
        pool.join()

        if self.progress:
            self.progress.finish()

def run(args):
    if args.download:
        resolvers = download_resolvers()
    else:
        resolvers = load_resolvers(args.resolvers)
    random.shuffle(resolvers)

    pool = gevent.pool.Pool(args.concurrency)

    bar = progressbar.ProgressBar(redirect_stdout=True, redirect_stderr=True)
    for resolver in bar(resolvers):
        pool.add(gevent.spawn(check_resolver, args, resolver))
    pool.join()

def _output_result(self, domain, name, query_type, result):
        """
        Output results, in various formats, to necessary places
        """
        # To console
        if name is None:
            dnsname = domain
        else:
            dnsname = '.'.join([name, domain])
        res_keys = ' '.join(['='.join([key, str(value)])
                             for key, value in result.items()])
        info = ' '.join([dnsname, query_type, res_keys])
        if not self.options.quiet:
            print(info)
        #
        # Shit out same as console, but to file
        output = self.options.output
        if output:
            output.write(info + "\n")
            output.flush()
        #
        # Optionally shit out JSON
        outjson = self.options.json
        if outjson:
            outdict = result.copy()
            outdict['_type'] = query_type
            outdict['_domain'] = domain
            outdict['_name'] = name
            outdict.update(self.options.extra)
            if name and name[0] == '*':
                outdict['_wildcard'] = True
            outjson.write(json.dumps(outdict) + "\n")
            outjson.flush()

def _dnsresp_to_dict(self, obj):
        """
        Converts DNS reponse into a normalised dictionary
        """
        rdtype = obj.rdtype
        if rdtype in (dns.rdatatype.A, dns.rdatatype.AAAA):
            return dict(host=obj.address)
        elif rdtype == dns.rdatatype.SOA:
            return dict(retry=obj.retry, serial=obj.serial, expires=obj.expire,
                        refresh=obj.refresh, minttl=obj.minimum,
                        hostmaster=str(obj.rname), nsname=str(obj.mname))
        elif rdtype == dns.rdatatype.NS:
            return dict(host=str(obj.target))
        elif rdtype == dns.rdatatype.MX:
            return dict(priority=obj.preference, host=str(obj.exchange))
        elif rdtype == dns.rdatatype.CNAME:
            return dict(cname=str(obj.target))
        elif rdtype in (dns.rdatatype.TXT, dns.rdatatype.SPF):
            return dict(text=" ".join(obj.strings))
        elif rdtype == dns.rdatatype.SRV:
            return dict(priority=obj.priority, host=str(obj.target), port=obj.port,
                        weight=obj.weight)
        elif rdtype == dns.rdatatype.DS:
            return dict(keytag=obj.key_tag, hashtype=obj.digest_type,
                        hash=hexlify(obj.digest))
        elif rdtype == dns.rdatatype.DLV:
            return dict(keytag=obj.key_tag, hashtype=obj.digest_type)
        elif rdtype == dns.rdatatype.DNSKEY:
            return dict(keytag=dns.dnssec.key_id(obj), protocol=obj.protocol,
                        flags=obj.flags, algorithm=obj.algorithm,
                        length=keylength(obj.algorithm, obj.key),
                        key=hexlify(obj.key))
        raise RuntimeError("Unknown DNS response type %r" % (obj,))
        #  'RRSIG', 'DS', 'DLV', 'DNSKEY', 'NSEC', 'NSEC3', 'NSEC3PARAM']
        # TODO: add DS, DLV, RRSIG, NSEC, NSEC3, PTR, DNSKEY, SSHFP, NAPTR

def _find_wildcards(self):
        """
        Queries some random non-existant records to reduce false positives.
        Returns True if process can continue, otherwise false.
        """
        wildcard_count = self.options.wildcard_tests
        if wildcard_count < 1:
            return True
        total_queries = len(self.domains) * wildcard_count
        LOG.info("Eliminating wildcard responses (%d tests)", total_queries)
        is_ok = False
        # Setup pool and progress
        pool = gevent.pool.Pool(self.options.concurrency)
        if self.progress:
            self.progress.start(total_queries)
        self.finished = 0
        try:
            for domain in self.domains:
                LOG.debug("Checking wildcard domain: %s", domain)
                names = [rand_name() for _ in range(0, wildcard_count)]
                for name in names:
                    pool.add(gevent.spawn(self._test_wildcard, domain, name))
            is_ok = True
        except KeyboardInterrupt:
            print("Ctrl+C caught... stopping")
        pool.join()
        if self.progress:
            self.progress.finish()
        return is_ok

def benchmark():
    gevent.spawn(printstats)
    for _ in xrange(1000):
        pool.spawn(bench, itemid) 
    pool.join()

def sync_cassandra(simple=False):
    for table, fields in sorted(schemas.items(), key=lambda x: len(x[0])):
        if simple:
            if table in ['ataobao2.top10', 'ataobao2.blacklist', 'ataobao2.agghosts', 'ataobao2.cate', 'ataobao2.brand']:
                sync_table(table, fields)
        else:
            sync_table(table, fields)
    pool.join()

def test_throttling():
    def printget():
        print time.time(), 'id' in get_item(22183623058)

    import gevent.pool
    pool = gevent.pool.Pool(20)
    while True:
        pool.spawn(call_with_throttling, printget, threshold_per_minute=600)
    pool.join()

def run(self):
        def on_ids(ids):
            ai2.put(*ids)
            ItemCT.add_items(*ids)

        for cid in fecids:
            self.pool.spawn(list_cat, cid, on_ids=on_ids, use_pool=False, num_paths=1, max_page=1)
        self.pool.join()

def work(self):
        def workon(iap):
            pool = gevent.pool.Pool(self.poolsize)

            for i in range(self.poolsize):
                pool.spawn(iap.work)

            pool.join()

        gevent.joinall([gevent.spawn(workon, p) for p in self.processes])

def _run_items(mode, items, session, workers=None):
    ''' Multiprocess is not compatible with Windows !!! '''
    if mode == "mproc":
        '''Using ThreadPoolExecutor as managers to control the lifecycle of processes.
        Each thread will spawn a process and terminates when the process joins.
        '''
        def run_task_in_proc(item, index):
            proc = multiprocessing.Process(target=_run_next_item, args=(session, item, index))
            proc.start()
            proc.join()

        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
            for index, item in enumerate(items):
                executor.submit(run_task_in_proc, item, index)

    elif mode == "mthread":
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
            for index, item in enumerate(items):
                executor.submit(_run_next_item, session, item, index)

    elif mode == "asyncnet":
        import gevent
        import gevent.monkey
        import gevent.pool
        gevent.monkey.patch_all()
        pool = gevent.pool.Pool(size=workers)
        for index, item in enumerate(items):
            pool.spawn(_run_next_item, session, item, index)
        pool.join()

    else:
        for i, item in enumerate(items):
            nextitem = items[i + 1] if i + 1 < len(items) else None
            item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
            if session.shouldstop:
                raise session.Interrupted(session.shouldstop)

def wait_for_res(self):
        while not self.res_is_downloaded:
            if not self.download_thread.is_alive():
                logger.info('download thread exit in exceptions')
                self.download_thread.join()
                # raise DownloadError('download thread exit in exceptions')
            logger.debug(
                "Waiting for resource to complete, "
                "size of %s is %s now" %
                (self.res_url, self.res_downloaded_size()))
            time.sleep(2)
        return True

def check_download_thread(self):
        if self.download_thread.is_alive():
            return True
        else:
            self.download_thread.join()

def on_result(self, url, resp, extra=None):
        status = dict(
            url=resp.url or url,
            hist=[(hist.status_code, hist.url) for hist in resp.history],
            sc=resp.status_code,
            hds=[K for K in resp.headers],
            cks=[C.name for C in resp.cookies],
            hd={k: v for k, v in dict(
                lm=resp.headers.get('Last-Modified'),
                ct=resp.headers.get('Content-Type'),
                cl=resp.headers.get('Content-Length'),
                sv=resp.headers.get('Server'),
            ).iteritems() if v}
        )
        if extra and isinstance(extra, dict):
            status.update(extra)
        # Save file to storage
        storage = self.options.storage
        if storage:
            url_hash = sha1_b32(status['url'],
                                resp.status_code,
                                resp.headers.get('Last-Modified'),
                                resp.headers.get('Date'),
                                resp.headers.get('Content-Len'))[:12]
            url_dir = os.path.join(storage, url_hash[1])
            url_path = os.path.join(url_dir, url_hash[1:])
            os.makedirs(url_dir)
            with open(url_path, 'wb') as handle:
                for chunk in resp.iter_content(chunk_size=1024*64):
                    handle.write(chunk)
            status['id'] = url_hash
        self._log_result(status)

def sync_table(table, fields):
    f1 = ', '.join(fields)
    pieces = {
        'ataobao2.item': 100,
        'ataobao2.item_by_date': 1000,
        'ataobao2.brand_by_date': 10,
        'ataobao2.shop_by_date': 10,
    }.get(table, 1)
    start = -2**63
    step = 2**64/pieces
    print 'migrating {} {}'.format(table, f1)

    for i in range(pieces):
        start = -2**63 + step*i
        end = min(2**63-1, -2**63+step*(i+1))
        with db1.connection() as cur:
            print 'piece', i+1
            #print 'select {} from {} where token({})>=:v1 and token({})<:v2'.format(f1, table, fields[0], fields[0]), dict(v1=start, v2=end)
            if table.endswith('_by_date') and 'datestr' in fields:
                d0 = (datetime.utcnow() + timedelta(hours=8) - timedelta(days=2)).strftime('%Y-%m-%d')
                cur.execute('select {} from {} where token({})>=:v1 and token({})<:v2 and datestr>=:d0 allow filtering'.format(f1, table, fields[0], fields[0]), 
                        dict(v1=start, v2=end, d0=d0), consistency_level='ONE')
            else:
                cur.execute('select {} from {} where token({})>=:v1 and token({})<:v2'.format(f1, table, fields[0], fields[0]), 
                        dict(v1=start, v2=end), consistency_level='ONE')
            for j, row in enumerate(cur):
                if j % 1000 == 0:
                    print 'syncd {}'.format(j)
                params = {}
                fs = list(fields)
                for k,v in zip(fields, row):
                    if k == 'date':
                        if v is not None and len(v)==8:
                            v = struct.unpack('!q', v)[0]
                        else:
                            continue
                    if v is not None:
                        params[k] = v 
                fs = params.keys()
                fs1 = ', '.join(fs)
                fs2 = ', '.join([':'+f for f in fs])
                if 'id' in params or 'datestr' in params or 'name' in params:
                    if table == 'ataobao2.item_by_date' and 'date' not in params:
                        continue
                    #print 'INSERT INTO {} ({}) VALUES ({})'.format(table, fs1, fs2), params
                    pool.spawn(db2.execute, 'insert into {} ({}) values ({})'.format(table, fs1, fs2), params)