def test_basic(self):
        yappi.set_clock_type('wall')
        def dummy():
            pass
        def a():
            time.sleep(0.2)
        class Worker1(threading.Thread):
            def a(self):
                time.sleep(0.3)
            def run(self):
                self.a()
        yappi.start(builtins=False, profile_threads=True)

        c = Worker1()
        c.start()
        c.join()
        a()
        stats = yappi.get_func_stats()
        fsa1 = utils.find_stat_by_name(stats, 'Worker1.a')
        fsa2 = utils.find_stat_by_name(stats, 'a')
        self.assertTrue(fsa1 is not None)
        self.assertTrue(fsa2 is not None)
        self.assertTrue(fsa1.ttot > 0.2)
        self.assertTrue(fsa2.ttot > 0.1)
        tstats = yappi.get_thread_stats()
        self.assertEqual(len(tstats), 2)
        tsa = utils.find_stat_by_name(tstats, 'Worker1')
        tsm = utils.find_stat_by_name(tstats, '_MainThread')
        dummy() # call dummy to force ctx name to be retrieved again.
        self.assertTrue(tsa is not None)
        # TODO: I put dummy() to fix below, remove the comments after a while.
        self.assertTrue( # FIX: I see this fails sometimes?
            tsm is not None, 
            'Could not find "_MainThread". Found: %s' % (', '.join(utils.get_stat_names(tstats)))) 

def main() -> None:
    '''Runs server'''

    # Parse options
    define('production',
               default = False,
               help = 'run in production mode',
               type = bool)
    options.parse_command_line()

    # Set server name
    pname = settings.process_name if settings.process_name else None
    if pname:
        setproctitle(pname)

    # Register IRC server
    server = IRCServer(settings = ircdsettings)
    for address, port in ircdsettings['listen']:
        server.listen(port, address = address)

    # Start profiling
    if settings.profiling:
        import yappi
        yappi.start()

    # Setup autoreload
    autoreload.start()

    # Run application
    IOLoop.instance().start()

def main4():
    import yappi
    yappi.start()
    main2()
    stats = yappi.get_func_stats()
    stats.sort('ttot')
    stats.print_all()

    def __init__(self, port, num_workers):
        self._workers = {}
        self.num_workers = num_workers
        self._port = port
        self._server = rpc.listen("0.0.0.0", port)
        self._server.register_object(self)
        self._initialized = False
        self._server.serve_nonblock()
        self._ctx = None

        self._worker_statuses = {}
        self._worker_scores = {}
        self._available_workers = []

        self._arrays = weakref.WeakSet()

        if FLAGS.profile_master:
            import yappi

            yappi.start()
            atexit.register(_dump_profile)
        atexit.register(_shutdown)

        global MASTER
        MASTER = self

    def test_print_formatting(self):
        def a():
            pass

        def b():
            a()

        func_cols = {1: ("name", 48), 0: ("ncall", 5), 2: ("tsub", 8)}
        thread_cols = {1: ("name", 48), 0: ("ttot", 8)}

        yappi.start()
        a()
        b()
        yappi.stop()
        fs = yappi.get_func_stats()
        cs = fs[1].children
        ts = yappi.get_thread_stats()
        # fs.print_all(out=sys.stderr, columns={1:("name", 70), })
        # cs.print_all(out=sys.stderr, columns=func_cols)
        # ts.print_all(out=sys.stderr, columns=thread_cols)
        # cs.print_all(out=sys.stderr, columns={})

        self.assertRaises(yappi.YappiError, fs.print_all, columns={1: ("namee", 9)})
        self.assertRaises(yappi.YappiError, cs.print_all, columns={1: ("dd", 0)})
        self.assertRaises(yappi.YappiError, ts.print_all, columns={1: ("tidd", 0)})

 def test_merge_multithreaded_stats(self):
     import threading
     import _yappi
     timings = {"a_1":2, "b_1":1}
     _yappi._set_test_timings(timings)
     def a(): pass
     def b(): pass
     yappi.start()
     t = threading.Thread(target=a)
     t.start()
     t.join()
     t = threading.Thread(target=b)
     t.start()
     t.join()
     yappi.get_func_stats().save("ystats1.ys")
     yappi.clear_stats()
     _yappi._set_test_timings(timings)
     self.assertEqual(len(yappi.get_func_stats()), 0)
     self.assertEqual(len(yappi.get_thread_stats()), 1)
     t = threading.Thread(target=a)
     t.start()
     t.join()
     
     self.assertEqual(_yappi._get_start_flags()["profile_builtins"], 0)
     self.assertEqual(_yappi._get_start_flags()["profile_multithread"], 1)
     yappi.get_func_stats().save("ystats2.ys")
    
     stats = yappi.YFuncStats(["ystats1.ys", "ystats2.ys",])
     fsa = utils.find_stat_by_name(stats, "a")
     fsb = utils.find_stat_by_name(stats, "b")
     self.assertEqual(fsa.ncall, 2)
     self.assertEqual(fsb.ncall, 1)
     self.assertEqual(fsa.tsub, fsa.ttot, 4)
     self.assertEqual(fsb.tsub, fsb.ttot, 1)

    def test_producer_consumer_with_queues(self):
        # we currently just stress yappi, no functionality test is done here.
        yappi.start()
        import time
        if utils.is_py3x():
            from queue import Queue
        else:
            from Queue import Queue
        from threading import Thread
        WORKER_THREAD_COUNT = 50
        WORK_ITEM_COUNT = 2000
        def worker():
            while True:
                item = q.get()                
                # do the work with item
                q.task_done()

        q = Queue()
        for i in range(WORKER_THREAD_COUNT):
            t = Thread(target=worker)
            t.daemon = True
            t.start()
             
        for item in range(WORK_ITEM_COUNT):
            q.put(item)
        q.join()# block until all tasks are done
        #yappi.get_func_stats().sort("callcount").print_all()
        yappi.stop()

 def start(self):
     yappi.set_context_id_callback(lambda:id(greenlet.getcurrent()))
     yappi.set_context_name_callback(lambda:greenlet.getcurrent().__class__.__name__)
     yappi.set_clock_type('cpu')
     yappi.start(builtins=True)
     self._isStart = True
     return 'success'

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--host', default=None,  # NOT -h, that's help
                    help='host that coordinated will listen on, '
                    '0.0.0.0 for any input interface')
    ap.add_argument('--port', '-p', type=int, default=None,
                    help='port number that coordinated will listen on')
    ap.add_argument('--pid', default=None,
                    help='file to write pid to')
    ap.add_argument('--snapshot-dir', default=None,
                    help='direcotry to write snapshots to')
    ap.add_argument('--httpd', default=None,
                    help='ip:port or :port to serve http info on')
    if yappi is not None:
        ap.add_argument('--yappi', default=None, help='file to write yappi profiling to. will be suffied by {timestamp}.txt')
    args = yakonfig.parse_args(ap, [yakonfig, dblogger, coordinate])

    if args.pid:
        with open(args.pid, 'w') as f:
            f.write(str(os.getpid()))

    if args.snapshot_dir is not None:
        cjqconfig = yakonfig.get_global_config('coordinate', 'job_queue')
        # (This modifies the global configuration in place)
        cjqconfig['snapshot_path_format'] = os.path.join(
            args.snapshot_dir, 'snapshot_{timestamp}')

    if (yappi is not None) and args.yappi:
        yappi.start()
        yt = threading.Thread(target=yappi_logger, args=(args.yappi,))
        yt.daemon = True
        yt.start()

    daemon = CoordinateServer(host=args.host, port=args.port, httpd=args.httpd)
    daemon.run()

    def test_singlethread_profiling(self):
        yappi.set_clock_type("wall")

        def a():
            time.sleep(0.2)

        class Worker1(threading.Thread):
            def a(self):
                time.sleep(0.3)

            def run(self):
                self.a()

        yappi.start(profile_threads=False)

        c = Worker1()
        c.start()
        c.join()
        a()
        stats = yappi.get_func_stats()
        fsa1 = utils.find_stat_by_name(stats, "Worker1.a")
        fsa2 = utils.find_stat_by_name(stats, "a")
        self.assertTrue(fsa1 is None)
        self.assertTrue(fsa2 is not None)
        self.assertTrue(fsa2.ttot > 0.1)

    def test_basic(self):
        import threading
        import time
        yappi.set_clock_type('wall')
        def a():
            time.sleep(0.2)
        class Worker1(threading.Thread):
            def a(self):
                time.sleep(0.3)                
            def run(self):
                self.a()
        yappi.start(builtins=False, profile_threads=True)

        c = Worker1()
        c.start()
        c.join()        
        a()
        stats = yappi.get_func_stats()
        fsa1 = utils.find_stat_by_name(stats, 'Worker1.a')
        fsa2 = utils.find_stat_by_name(stats, 'a')
        self.assertTrue(fsa1 is not None)
        self.assertTrue(fsa2 is not None)
        self.assertTrue(fsa1.ttot > 0.2)
        self.assertTrue(fsa2.ttot > 0.1)
        tstats = yappi.get_thread_stats()
        self.assertEqual(len(tstats), 2)
        tsa = utils.find_stat_by_name(tstats, 'Worker1')
        tsm = utils.find_stat_by_name(tstats, '_MainThread')
        self.assertTrue(tsa is not None)
        self.assertTrue(tsm is not None) # FIX: I see this fails sometimes?

 def test_children_stat_functions(self):
     _timings = {"a_1":5, "b_1":3, "c_1":1}
     _yappi._set_test_timings(_timings)
     def b(): 
         pass
     def c():
         pass
     def a():
         b()
         c()
     yappi.start()
     a()
     b() # non-child call
     c() # non-child call
     stats = yappi.get_func_stats()
     fsa = utils.find_stat_by_name(stats, 'a')
     childs_of_a = fsa.children.get().sort("tavg", "desc")
     prev_item = None
     for item in childs_of_a:
         if prev_item:
             self.assertTrue(prev_item.tavg > item.tavg)
         prev_item = item
     childs_of_a.sort("name", "desc")
     prev_item = None
     for item in childs_of_a:
         if prev_item:
             self.assertTrue(prev_item.name > item.name)
         prev_item = item
     childs_of_a.clear()
     self.assertTrue(childs_of_a.empty())

    def test_callback(self):
        self.context_id = 0
        yappi.set_context_id_callback(lambda: self.context_id)
        yappi.start()
        a()
        self.context_id = 1
        a()
        self.context_id = 2
        a()

        # Re-schedule context 1.
        self.context_id = 1
        a()
        yappi.stop()

        threadstats = yappi.get_thread_stats().sort('id', 'ascending')
        self.assertEqual(3, len(threadstats))
        self.assertEqual(0, threadstats[0].id)
        self.assertEqual(1, threadstats[1].id)
        self.assertEqual(2, threadstats[2].id)

        self.assertEqual(1, threadstats[0].sched_count)
        self.assertEqual(2, threadstats[1].sched_count)  # Context 1 ran twice.
        self.assertEqual(1, threadstats[2].sched_count)

        funcstats = yappi.get_func_stats()
        self.assertEqual(4, utils.find_stat_by_name(funcstats, 'a').ncall)

    def test_profile_single_context(self):
        
        def id_callback():
            return self.callback_count
        def a():
            pass

        self.callback_count = 1
        yappi.set_context_id_callback(id_callback)
        yappi.start(profile_threads=False)
        a() # context-id:1
        self.callback_count = 2
        a() # context-id:2
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, "a")
        self.assertEqual(fsa.ncall, 1)
        yappi.stop()
        yappi.clear_stats()
        
        self.callback_count = 1
        yappi.start() # profile_threads=True
        a() # context-id:1
        self.callback_count = 2
        a() # context-id:2
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, "a")
        self.assertEqual(fsa.ncall, 2)

    def test_pause_resume(self):
        yappi.set_context_id_callback(lambda: self.context_id)
        yappi.set_clock_type('wall')

        # Start in context 0.
        self.context_id = 0
        yappi.start()
        time.sleep(0.08)

        # Switch to context 1.
        self.context_id = 1
        time.sleep(0.05)

        # Switch back to context 0.
        self.context_id = 0
        time.sleep(0.07)

        yappi.stop()

        t_stats = yappi.get_thread_stats().sort('id', 'ascending')
        self.assertEqual(2, len(t_stats))
        self.assertEqual(0, t_stats[0].id)
        self.assertEqual(2, t_stats[0].sched_count)
        self.assertTrue(0.15 < t_stats[0].ttot < 0.3)

        self.assertEqual(1, t_stats[1].id)
        self.assertEqual(1, t_stats[1].sched_count)
        # Context 1 was first scheduled 0.08 sec after context 0.
        self.assertTrue(0.1 < t_stats[1].ttot < 0.2 )

 def test_yappi_overhead(self):  
     import time
     LOOP_COUNT = 10000
     def a(): pass
     def b():
         for i in range(LOOP_COUNT): a()
     t0 = time.time()
     yappi.start()
     b()
     yappi.stop()
     time_with_yappi = time.time() - t0
     t0 = time.time()
     b()
     time_without_yappi = time.time() - t0
     if time_without_yappi == 0:
         time_without_yappi = 0.000001
         
     # in latest v0.82, I calculated this as close to "7.0" in my machine.
     # however, %83 of this overhead is coming from tickcount(). The other %17
     # seems to have been evenly distributed to the internal bookkeeping 
     # structures/algorithms which seems acceptable. Note that our test only 
     # tests one function being profiled at-a-time in a short interval. 
     # profiling high number of functions in a small time
     # is a different beast, (which is pretty unlikely in most applications)
     # So as a conclusion: I cannot see any optimization window for Yappi that
     # is worth implementing as we will only optimize %17 of the time.
     sys.stderr.write("\r\nYappi puts %0.1f times overhead to the profiled application in average.\r\n" % \
         (time_with_yappi / time_without_yappi))        

    def test_filter(self):
        def a(): pass
        def b(): a()
        def c(): b()

        _TCOUNT = 5

        ts = []
        yappi.start()
        for i in range(_TCOUNT):
            t = threading.Thread(target=c)
            t.start()
            ts.append(t)

        for t in ts:
            t.join()

        yappi.stop()

        fstats = yappi.get_func_stats(filter={"ctx_id":9})
        self.assertTrue(fstats.empty())
        fstats = yappi.get_func_stats(filter={"ctx_id":0, "name":"c"}) # main thread
        self.assertTrue(fstats.empty())

        for i in range(1, _TCOUNT):
            fstats = yappi.get_func_stats(filter={"ctx_id":i, "name":"a", 
                "ncall":1})
            self.assertEqual(fstats.pop().ncall, 1)
            fstats = yappi.get_func_stats(filter={"ctx_id":i, "name":"b"})
            self.assertEqual(fstats.pop().ncall, 1)
            fstats = yappi.get_func_stats(filter={"ctx_id":i, "name":"c"})
            self.assertEqual(fstats.pop().ncall, 1)

 def get(self):
     self.interval = int(self.get_argument('interval', self.interval))
     self.sort_type = yappi.SORT_TYPES_FUNCSTATS[self.get_argument(
         'sort_type', 'tsub')]
     self.limit = self.get_argument('limit', self.limit)
     yappi.start()
     tornado.ioloop.IOLoop.instance().add_timeout(
         datetime.timedelta(seconds=self.interval), self.finish_profile)

 def test_lambda(self):
     f = lambda: time.sleep(0.3)
     yappi.set_clock_type("wall")
     yappi.start()
     f()
     stats = yappi.get_func_stats()
     fsa = utils.find_stat_by_name(stats, "<lambda>")
     self.assertTrue(fsa.ttot > 0.1)

 def test_concurrent_futures(self):
     yappi.start()
     from concurrent.futures import ThreadPoolExecutor
     with ThreadPoolExecutor(max_workers=5) as executor:
         f = executor.submit(pow, 5, 2)
         self.assertEqual(f.result(), 25)        
     time.sleep(1.0)
     yappi.stop()