def _check_executor(self, dt):
        start = time()
        try:
            for future in as_completed(self._futures[:], 0):
                self._futures.remove(future)
                try:
                    result = future.result()
                except Exception:
                    traceback.print_exc()
                    # make an error tile?
                    continue
                if result is None:
                    continue
                callback, args = result
                callback(*args)

                # capped executor in time, in order to prevent too much
                # slowiness.
                # seems to works quite great with big zoom-in/out
                if time() - start > self.cap_time:
                    break
        except TimeoutError:
            pass 

def count(self, lines):
        # use the name server's prefix lookup to get all registered wordcounters
        with locate_ns() as ns:
            all_counters = ns.list(prefix="example.dc2.wordcount.")

        # chop the text into chunks that can be distributed across the workers
        # uses futures so that it runs the counts in parallel
        # counter is selected in a round-robin fashion from list of all available counters
        with futures.ThreadPoolExecutor() as pool:
            roundrobin_counters = cycle(all_counters.values())
            tasks = []
            for chunk in grouper(200, lines):
                tasks.append(pool.submit(self.count_chunk, next(roundrobin_counters), chunk))

            # gather the results
            print("Collecting %d results (counted in parallel)..." % len(tasks))
            totals = Counter()
            for task in futures.as_completed(tasks):
                try:
                    totals.update(task.result())
                except Pyro5.errors.CommunicationError as x:
                    raise Pyro5.errors.PyroError("Something went wrong in the server when collecting the responses: "+str(x))
            return totals 

def run(self, concurrent=10):
        """
        Entry point.
        :param concurrent: number of threads to use
        :return: message json
        """

        children = [self.stac_file]
        logger.info(f"Using {concurrent} threads")
        while True:
            with futures.ThreadPoolExecutor(max_workers=int(concurrent)) as executor:
                future_tasks = [executor.submit(self._validate, url) for url in children]
                children = []
                for task in futures.as_completed(future_tasks):
                    message, status, new_children = task.result()
                    self.status = self._update_status(self.status, status)
                    self.message.append(message)
                    children.extend(new_children)

            if not children:
                break

        return json.dumps(self.message) 

def download_many(cc_list):
    cc_list = cc_list[:5]  # <1>
    with futures.ThreadPoolExecutor(max_workers=3) as executor:  # <2>
        to_do = []
        for cc in sorted(cc_list):  # <3>
            future = executor.submit(download_one, cc)  # <4>
            to_do.append(future)  # <5>
            msg = 'Scheduled for {}: {}'
            print(msg.format(cc, future))  # <6>

        results = []
        for future in futures.as_completed(to_do):  # <7>
            res = future.result()  # <8>
            msg = '{} result: {!r}'
            print(msg.format(future, res))  # <9>
            results.append(res)

    return len(results) 

def runner(k):
        threadpool = thread.ThreadPoolExecutor(max_workers=k.get('threads'))
        if k.get('verbose'):
            info('Set %s threads..'%k.get('threads'))
        futures = (threadpool.submit(requester,domain,k.get("proxy"),k.get("timeout"),
                k.get("output"),k.get('process'),k.get('verbose')) for domain in k.get("domains"))
        for i,results in enumerate(thread.as_completed(futures)):
            if k.get('verbose') and k.get('d_list'):
                str_ = "{i}{b:.2f}% Domain: {d}".format(
                    i=_info(),
                    b=PERCENT(int(i),
                        int(k.get('dict_len'))),d=k.get('domains')[i]
                    )
                print_(str_)
            else:
                info('Domain: {}'.format(k.get('domains')[i]))
            pass 

def par_crop(args):
    """
    Dataset curation,crop data and transform the format of a label
    """
    crop_path = os.path.join(args.download_dir, './crop{:d}'.format(args.instance_size))
    if not os.path.isdir(crop_path): makedirs(crop_path)
    VID_base_path = os.path.join(args.download_dir, './ILSVRC')
    ann_base_path = os.path.join(VID_base_path, 'Annotations/DET/train/')
    sub_sets = ('a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i')
    for sub_set in sub_sets:
        sub_set_base_path = os.path.join(ann_base_path, sub_set)
        if 'a' == sub_set:
            xmls = sorted(glob.glob(os.path.join(sub_set_base_path, '*', '*.xml')))
        else:
            xmls = sorted(glob.glob(os.path.join(sub_set_base_path, '*.xml')))
        n_imgs = len(xmls)
        sub_set_crop_path = os.path.join(crop_path, sub_set)
        with futures.ProcessPoolExecutor(max_workers=args.num_threads) as executor:
            fs = [executor.submit(crop_xml, args, xml, sub_set_crop_path, args.instance_size) for xml in xmls]
            for i, f in enumerate(futures.as_completed(fs)):
                printProgress(i, n_imgs, prefix=sub_set, suffix='Done ', barLength=80) 

def par_crop(args, ann_base_path):
    """
    Dataset curation, crop data and transform the format of label
    Parameters
    ----------
    ann_base_path: str, Annotations base path
    """
    crop_path = os.path.join(args.download_dir, './crop{:d}'.format(int(args.instance_size)))
    if not os.path.isdir(crop_path):
        makedirs(crop_path)
    sub_sets = sorted({'a', 'b', 'c', 'd', 'e'})
    for sub_set in sub_sets:
        sub_set_base_path = os.path.join(ann_base_path, sub_set)
        videos = sorted(os.listdir(sub_set_base_path))
        n_videos = len(videos)
        with futures.ProcessPoolExecutor(max_workers=args.num_threads) as executor:
            fs = [executor.submit(crop_video, args, sub_set, video, crop_path, ann_base_path) for video in videos]
            for i, f in enumerate(futures.as_completed(fs)):
                # Write progress to error so that it can be seen
                printProgress(i, n_videos, prefix=sub_set, suffix='Done ', barLength=40) 

def _visual_items_upload_with_operation(self, sequence, visual_item_upload_operation):
        items_to_upload = []
        for visual_item in sequence.visual_items:
            if str(visual_item.index) not in sequence.progress:
                items_to_upload.append(visual_item)

        with THREAD_LOCK:
            self.manager.progress_bar.update(len(sequence.visual_items) - len(items_to_upload))

        with ThreadPoolExecutor(max_workers=self.workers) as executor:
            future_events = [executor.submit(visual_item_upload_operation.upload,
                                             visual_item) for visual_item in items_to_upload]
            for completed_event in as_completed(future_events):
                uploaded, index = completed_event.result()
                with THREAD_LOCK:
                    if uploaded:
                        self.__persist_upload_index(index, sequence.path)
                        sequence.progress.append(index)
                    self.manager.progress_bar.update(1) 

def test_temp_table_concurrency(con, test_data_dir):
    # we don't install futures on windows in CI and we can't run this test
    # there anyway so we import here
    import concurrent.futures
    from concurrent.futures import as_completed

    def limit_10(i, hdfs_path):
        t = con.parquet_file(hdfs_path)
        return t.sort_by(t.r_regionkey).limit(1, offset=i).execute()

    nthreads = 4
    hdfs_path = pjoin(test_data_dir, 'parquet/tpch_region')

    with concurrent.futures.ThreadPoolExecutor(max_workers=nthreads) as e:
        futures = [e.submit(limit_10, i, hdfs_path) for i in range(nthreads)]
    assert all(map(len, (future.result() for future in as_completed(futures)))) 

def test_multiple(ActorClass):
    print('-----------------')
    print('Test multiple for {}'.format(ActorClass))
    # Make multiple actors and send them each multiple jobs
    n_actors = 5
    n_jobs = 10
    actors_exs = [ActorClass.executor(a) for a in range(1, n_actors)]
    fs = []
    for jobid in range(n_jobs):
        n = jobid + 500
        fs += [ex.post({'action': 'prime', 'n': n}) for ex in actors_exs]

    for f in futures.as_completed(fs):
        print('n, a, prime = {}'.format(f.result()))

    actors = [ex.post({'action': 'debug'}).result() for ex in actors_exs]
    for a in actors:
        print(a.state)
    print('Test completed')
    print('L______________') 

def process(self, start_url, crawled_urls):
        self.output.info("Checking ldap injection...")
        db = self.datastore.open("ldap.txt", "r")
        dbfiles = [x.strip() for x in db]

        for payload in dbfiles:
            with ThreadPoolExecutor(max_workers=None) as executor:
                futures = [
                    executor.submit(self.attack, payload, url) for url in crawled_urls
                ]
        try:
            for future in as_completed(futures):
                future.result()
        except KeyboardInterrupt:
            executor.shutdown(False)
            raise 

def process(self, start_url, crawled_urls):
        self.output.info("Checking common backup files..")
        db = self.datastore.open("bfile.txt", "r")
        dbfiles = [x.strip() for x in db.readlines()]
        db1 = self.datastore.open("cfile.txt", "r")
        dbfiles1 = [x.strip() for x in db1.readlines()]
        urls = []
        for b in dbfiles:
            for d in dbfiles1:
                bdir = b.replace("[name]", d)
                urls.append(urljoin(str(start_url), str(bdir)))
        # We launch ThreadPoolExecutor with max_workers to None to get default optimization
        # https://docs.python.org/3/library/concurrent.futures.html
        with ThreadPoolExecutor(max_workers=None) as executor:
            futures = [executor.submit(self.check_url, url) for url in urls]
            try:
                for future in as_completed(futures):
                    future.result()
            except KeyboardInterrupt:
                executor.shutdown(False)
                raise 

def process(self, start_url, crawled_urls):
        self.output.info("Checking admin interfaces...")
        with self.datastore.open("admin.txt", "r") as db:
            dbfiles = [x.strip() for x in db.readlines()]
            urls = map(
                lambda adminpath: urljoin(str(start_url), str(adminpath)), dbfiles
            )
            # We launch ThreadPoolExecutor with max_workers to None to get default optimization
            # https://docs.python.org/3/library/concurrent.futures.html
            with ThreadPoolExecutor(max_workers=None) as executor:
                futures = [executor.submit(self.check_url, url) for url in urls]
                try:
                    for future in as_completed(futures):
                        future.result()
                except KeyboardInterrupt:
                    executor.shutdown(False)
                    raise 

def async_get(self, name: str, tag: str, parties: list) -> typing.Generator:
        rubbish = Rubbish(name, tag)
        futures = self._check_get_status_async(name, tag, parties)
        for future in as_completed(futures):
            party = futures[future]
            obj, head, frags = future.result()
            if isinstance(obj, _DTable):
                rubbish.add_table(obj)
                yield (party, obj)
            else:
                table, key = head
                rubbish.add_obj(table, key)
                if not is_split_head(obj):
                    yield (party, obj)
                else:
                    frag_table, frag_keys = frags
                    rubbish.add_table(frag_table)
                    fragments = [frag_table.get(key) for key in frag_keys]
                    yield (party, split_get(fragments))
        yield (None, rubbish) 

def test_zero_timeout(self):
        future1 = self.executor.submit(time.sleep, 2)
        completed_futures = set()
        try:
            for future in futures.as_completed(
                    [CANCELLED_AND_NOTIFIED_FUTURE,
                     EXCEPTION_FUTURE,
                     SUCCESSFUL_FUTURE,
                     future1],
                    timeout=0):
                completed_futures.add(future)
        except futures.TimeoutError:
            pass

        self.assertEqual(set([CANCELLED_AND_NOTIFIED_FUTURE,
                              EXCEPTION_FUTURE,
                              SUCCESSFUL_FUTURE]),
                         completed_futures) 

def run(self):
        self._running = True

        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            for file in self.files.keys():
                gain = GstGain(file, self.ref_level)
                self._futures[executor.submit(gain.gain)] = gain

            for future in futures_completed(self._futures):
                if self._running:
                    try:
                        self._post_process(*future.result())
                    except Exception:
                        # Call with the value stored in the GstGain object
                        self._post_process(*self._futures[future].result)
                else:
                    break

        if self._running:
            MainActionsHandler.do_action(self._action)
        else:
            logging.info('REPLY-GAIN:: Stopped by user')

        self.on_progress.emit(-1)
        self.on_progress.disconnect() 

def fetch(self):
        try:
            if not os.path.exists(self.target_dir):
                os.mkdir(self.target_dir)
        except Exception as e:
            print(e)
        self.page = urllib2.urlopen(self.base_url)
        self.data = BeautifulSoup(self.page.read(), "lxml")
        if not self.flag:
            table = self.data.findAll("table")[0]
            all_a = table.findAll("a")
            member_a = table.findAll("a", class_="tc_coder coder")
            all_set = set(all_a)
            member_set = set(member_a)
            post = list(set(all_set).difference(member_set))
        else:
            post = [self.base_url]

        with ThreadPoolExecutor(max_workers=4) as executor:
            future_to_url = {
                executor.submit(self.download, url): url for url in post}
            for future in as_completed(future_to_url):
                url = future_to_url[future] 

def test_zero_timeout(self):
        future1 = self.executor.submit(time.sleep, 2)
        completed_futures = set()
        try:
            for future in futures.as_completed(
                    [CANCELLED_AND_NOTIFIED_FUTURE,
                     EXCEPTION_FUTURE,
                     SUCCESSFUL_FUTURE,
                     future1],
                    timeout=0):
                completed_futures.add(future)
        except futures.TimeoutError:
            pass

        self.assertEqual(set([CANCELLED_AND_NOTIFIED_FUTURE,
                              EXCEPTION_FUTURE,
                              SUCCESSFUL_FUTURE]),
                         completed_futures) 

def run(self, data, max=4):
        results = []
        with futures.ThreadPoolExecutor(max_workers=max) as executor:
            future_to_url = {}
            for i, payload in enumerate(data):
                payload['chrome_id'] = i
                future_to_url[executor.submit(self.run1, payload)] = payload
                # future_to_url[executor.submit(self.run1_core, payload, browser, begin_time)] = payload
            for future in futures.as_completed(future_to_url):
                url = future_to_url[future]
                try:
                    data = future.result()
                except Exception as exc:
                    print('%r generated an exception: %s' % (url, exc))
                else:
                    data['chrome_id'] = url['chrome_id']
                    results.append(data)

        sorted_results = sorted(results, key=lambda tup: tup['chrome_id'])
        return sorted_results 

def main():
  with ThreadPoolExecutor(max_workers=3) as executor:
    
    tasks = []
    for url in URLS:
      task = executor.submit(checkStatus, (url))
      tasks.append(task)

    for future in as_completed(tasks):
      printStatus(future.result()) 

def as_completed(self, timeout=None):
        end_time = None
        if timeout is not None:
            end_time = timeout + time.monotonic()
        done = set()
        while end_time is None or end_time > time.monotonic():
            fs = OrderedDict([(e._future, e) for e in self.expectations if e not in done])
            for f in as_completed(fs.keys(), timeout=timeout):
                yield fs[f]
                done.add(fs[f])
            if len(done) == len(self.expectations):
                break 

def getgroups(grouplist):
    groupfuture=[]
    print "getgroups"
    for groupid in grouplist:
        if isinstance(groupid,basestring) and groupid.startswith("https"):
            groupfuture.append(session.get(str(groupid)))
        else:
            groupfuture.append(session.get(grouplookupurl.format(groupid)))
    badlist=[]
    pbar = tqdm(total=len(grouplist))
    for groupdata in as_completed(groupfuture):
        if groupdata.result().status_code==200:
            itemjson=groupdata.result().json()
            item=itemjson.get('group_id')
            if int(item) in sdegrouplist:
                try:
                    connection.execute(invGroups.update().where(invGroups.c.groupID == literal_column(str(item))),
                               groupID=item,
                               groupName=itemjson['name'],
                               categoryID=itemjson.get('category_id',None),
                               published=itemjson.get('published',False),
                               )
                except:
                    pass
            else:
                    connection.execute(invGroups.insert(),
                               groupID=item,
                               groupName=itemjson['name'],
                               categoryID=itemjson.get('category_id',None),
                               published=itemjson.get('published',False),
                                )
        else:
            badlist.append(groupdata.result().url)
            print groupdata.result().url
        pbar.update(1)
    return badlist 

def draw_results(self):
        for task in futures.as_completed(self.tasks):
            y, pixeldata = task.result()
            self.img.put(pixeldata, (0, y))
            self.root.update()
        duration = time.time() - self.start_time
        print("Calculation took: %.2f seconds" % duration) 

def screen(self, start, width):
        dr = width / self.res_x
        di = dr*(self.res_x/self.res_y)
        di *= 0.8   # aspect ratio correction
        self.result = ["?"] * self.res_y
        servers = [BatchProxy(Proxy(uri)) for uri in self.mandels]
        with futures.ThreadPoolExecutor(max_workers=len(servers)*2) as pool:
            for i in range(self.res_y):
                server = servers[i % len(servers)]
                server.calc_line(start, self.res_x, i*di, dr, i)
            tasks = [pool.submit(server) for server in servers]
            for task in futures.as_completed(tasks):
                lines = task.result()
                for (linenr, line) in lines:
                    self.result[linenr] = line
        return "\n".join(self.result) 

def attack(self, users, passwords, threads=1):
        pool = ThreadPoolExecutor(threads)
        threads = []

        for password in passwords:
            for user in users:
                t = pool.submit(self._handle_user_password, user, password)
                threads.append(t)

        for f in as_completed(threads):
            try:
                f.result()
            except Exception as ex:
                logging.debug('Error trying %s:%s %s' % (ex.kerb_user, ex.kerb_password, ex)) 

def download_many(cc_list, base_url, verbose, concur_req):
    counter = collections.Counter()
    with futures.ThreadPoolExecutor(concur_req) as executor:
        to_do_map = {}
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc, base_url, verbose)
            to_do_map[future] = cc
        to_do_iter = futures.as_completed(to_do_map)
        if not verbose:
            to_do_iter = tqdm.tqdm(to_do_iter, total=len(cc_list))
        for future in to_do_iter:
            try:
                res = future.result()
            except requests.exceptions.HTTPError as exc:
                error_msg = 'HTTP {res.status_code} - {res.reason}'
                error_msg = error_msg.format(res=exc.response)
            except requests.exceptions.ConnectionError as exc:
                error_msg = 'Connection error'
            else:
                error_msg = ''
                status = res.status

            if error_msg:
                status = HTTPStatus.error
            counter[status] += 1
            if verbose and error_msg:
                cc = to_do_map[future]
                print('*** Error for {}: {}'.format(cc, error_msg))

    return counter 

def download_many(cc_list, base_url, verbose, concur_req):
    counter = collections.Counter()
    with futures.ThreadPoolExecutor(max_workers=concur_req) as executor:  # <6>
        to_do_map = {}  # <7>
        for cc in sorted(cc_list):  # <8>
            future = executor.submit(download_one, cc, base_url,
                                     verbose)  # <9>
            to_do_map[future] = cc  # <10>
        done_iter = futures.as_completed(to_do_map)  # <11>
        if not verbose:
            done_iter = tqdm.tqdm(done_iter, total=len(cc_list))  # <12>
        for future in done_iter:  # <13>
            try:
                res = future.result()  # <14>
            except requests.exceptions.HTTPError as exc:  # <15>
                error_msg = 'HTTP {res.status_code} - {res.reason}'
                error_msg = error_msg.format(res=exc.response)
            except requests.exceptions.ConnectionError as exc:
                error_msg = 'Connection error'
            else:
                error_msg = ''
                status = res.status

            if error_msg:
                status = HTTPStatus.error
            counter[status] += 1
            if verbose and error_msg:
                cc = to_do_map[future]  # <16>
                print('*** Error for {}: {}'.format(cc, error_msg))

    return counter 

def main():
    with futures.ThreadPoolExecutor() as executor:

        downloads = [executor.submit(get_file, name, 60)
                        for name in names()]

        for future in futures.as_completed(downloads):
            try:
                name, length = future.result()
            except error.HTTPError as exc:
                print(f'*** {exc} ({exc.url})')
            else:
                print(f'{length:9,d} bytes\t{name}') 

def downLoadImg(destPath,infoList,img_size,thred_number):
    lencl= len(destPath)-1
    if destPath[lencl] == '/':
        destPath = destPath[:-1]
    className = destPath.split('/')
    className =  className[len(className)-1]
    def process(info):
        url = info['url']
        ext = info['format']
        idx = info['idx']
        print(idx)
        savePath = join(destPath,className+ str(idx) + '.' + ext)
        check = Path(savePath)
        if not check.is_file():
            print('Downloading : {} th {}' .format(idx,className))
            start = time.clock()
            p = mul.Process(target = rq.urlretrieve, name='download',args=(url,savePath))
            p.start()
            p.join(20)
            if p.is_alive():
                print('Too longdownloading terminate')
                p.terminate()
                p.join()
                call(['rm','-rf',savePath])
            if p.exitcode == 1:
                print('fail')
                call(['rm','-rf',savePath])
            else:
                resizeImg(savePath,img_size)
        else:
            print('Already Downloaded')
    with futures.ThreadPoolExecutor(max_workers=thred_number) as worker:
        mapper = [worker.submit(process,info) for info in infoList ]
        for tmp in tqdm(futures.as_completed(mapper), total=len(mapper)):
            pass


#Source txt file 

def downLoadImg(destPath,infoList,img_size,thred_number):
    checkFile(destPath)
    lencl= len(destPath)-1
    if destPath[lencl] == '/':
        destPath = destPath[:-1]
    className = destPath.split('/')
    className =  className[len(className)-1]
    def process(info):
        url = info['url']
        ext = 'jpeg'
        idx = info['idx']
        print(idx)
        savePath = join(destPath,className+ str(idx) + '.' + ext)
        check = Path(savePath)
        if not check.is_file():
            print('Downloading : {} th {}' .format(idx,className))
            start = time.clock()
            p = mul.Process(target = rq.urlretrieve, name='download',args=(url,savePath))
            p.start()
            p.join(200)
            if p.is_alive():
                print('Too longdownloading terminate')
                p.terminate()
                p.join()
                call(['rm','-rf',savePath])
            if p.exitcode == 1:
                print('fail')
                call(['rm','-rf',savePath])
            resizeImg(savePath,img_size)
            checkValid(savePath)
        else:
            print('Already Downloaded')
    with futures.ThreadPoolExecutor(max_workers=thred_number) as worker:
        mapper = [worker.submit(process,info) for info in infoList ]
        for tmp in tqdm(futures.as_completed(mapper), total=len(mapper)):
            pass