class StatisticQueue:
    def __init__(self, stats):
        self._semaphore = Semaphore()
        self.result = {}
        self.stats = stats

    def write_result(self, data):
        self._semaphore.acquire()
        self.result.update(data)
        self._semaphore.release()

    def start_parse(self):
        self.stats.connect()
        self.stats.init_message_stack()
        func_to_start = [
            self.stats.get_top3_speakers,
            self.stats.get_most_frequent_youtube_video,
            self.stats.get_time_activity,
            self.stats.get_abusive_expressions,
        ]
        threads = []
        for func in func_to_start:
            thread = Thread(target=func, args=(self, ))
            threads.append(thread)
            thread.start()
        for t in threads:
            t.join()
        return self.result

 def __init__(self, n, name):
     self.n = n
     self.name = name
     self.count = 0
     self.mutex = Semaphore(1)
     self.turnstile = Semaphore(0)
     self.turnstile2 = Semaphore(1)

class BinarySemaphore:
    def __init__(self, initial):
        self.sem = Semaphore(initial)
    def wait(self):
        self.sem.acquire()
    def signal(self):
        self.sem.release()

    def test_descriptor_clock(self):
        """Tests the clock trigger as a descriptor."""
        semp = Semaphore(0)

        class Foo(ModuleBase):
            clk = Clock()

            def __init__(self):
                super().__init__()
                self.bar = 0
                self.clk.frequency = 100

            @activity(clk)
            async def bas(self):
                self.bar += 1
                if self.bar >= 5:
                    self.clk.stop()
                    semp.release()

        foo = Foo()
        t0 = time()
        foo.clk.start()
        self.assertTrue(semp.acquire(timeout=0.1))
        self.assertGreaterEqual(time() - t0, 0.05)
        self.assertEqual(foo.bar, 5)

class ObjKeeper(object):
    """
    每种资源
    """

    def __init__(self, max_size):
        self.lock = Semaphore(max_size)
        self.objs = deque()

    def pop(self):
        # 获取锁
        self.lock.acquire()

        try:
            return self.objs.popleft()
        except:
            # 代表外面要重新生成新的
            return None

    def push(self, obj):
        if obj:
            self.objs.append(obj)

        # 无论如何都要释放
        self.lock.release()

class Synchronized:
	def __init__(self):
		from threading import Semaphore
		self.__lock = Semaphore()
		self.__ownerThread = None
		classdict = self.__class__.__dict__
		for attr in classdict.get("__synchronized__", ()):
			try:
				method = classdict[attr]
				if callable(method):
					self.__dict__[attr] = CallHook(self, method)
				else:
					if VERBOSE: print "! Synchronized: Object is not callable: %s" % (attr,)
			except KeyError:
				if VERBOSE: print "! Synchronized: Method not found: %s" % (attr,)

	def releaseInstance(self):
		self.__ownerThread = None
		self.__lock.release()

	def acquireInstance(self):
		self.__lock.acquire()
		self.__ownerThread = currentThread()

	def ownerThread(self):
		return self.__ownerThread

def pmap(f, l, limit=None):
    """A parallel version of map, that preserves ordering.
    Example:
    >>> pmap(lambda x: x*x, [1,2,3])
    [1, 4, 9]
    >>> import time
    >>> t1 = time.clock()
    >>> null = pmap(lambda x: time.sleep(1), range(10), 3)
    >>> time.clock() - t1 > 0.001
    True
    """
    if limit:
        pool_semaphore = Semaphore(limit)
    else:
        pool_semaphore = None

    pool = []
    res = range(len(l))
    for i in range(len(l)):
        t = Thread(target=mapper, args=(f, l[i], res, i, pool_semaphore))
        pool.append(t)
        if limit:
            pool_semaphore.acquire()
        t.start()
    map(lambda x: x.join(), pool)
    return res

def recover_images(parser, destination):
    """Parse images and save them to <manga>/<chapter>/<image>."""
    urls = parser.parse()
    manga_path = os.path.join(destination, parser.title)
    ch_digits = len(str(len(urls)))
    for chapter, pages in urls:
        #Normalize chapter digits
        chapter = "0" * (ch_digits - len(str(chapter))) + str(chapter)
        chapter_path = os.path.join(manga_path, chapter)
        if not os.path.exists(chapter_path):
            os.makedirs(chapter_path)
        savers = list()
        logging.info('Saving Chapter %s to %s', chapter, chapter_path)

        pg_digits = len(str(len(pages)))
        sem = Semaphore(BaseParser.MAX_CONNECTIONS)
        for page, url in enumerate(pages, start=1):
            sem.acquire()
            #Normalize page digits
            page = "0" * (pg_digits - len(str(page))) + str(page)
            path = os.path.join(chapter_path, str(page) + '.jpg')
            saver = utils.ImageSaver(path, url, sem)
            savers.append(saver)
            saver.start()
        map(lambda thread: thread.join(), savers)

class Footman:
    def __init__(self, num_philosophers, num_meals):
        self.num_philosophers = num_philosophers
        self.num_meals = num_meals
        self.forks = [Semaphore(1) for i in range(self.num_philosophers)]
        self.footman = Semaphore(self.num_philosophers - 1)  # at most one philosopher cannot dine

    def left(self, i):
        return i

    def right(self, i):
        return (i + 1) % self.num_philosophers

    def get_forks(self, i):
        self.footman.acquire()
        self.forks[self.right(i)].acquire()
        self.forks[self.left(i)].acquire()

    def put_forks(self, i):
        self.forks[self.right(i)].release()
        self.forks[self.left(i)].release()
        self.footman.release()

    def philosopher(self, id):
        while self.num_meals > 0:
            self.get_forks(id)
            # eating
            self.num_meals -= 1
            sleep(rng.random() / 100)
            self.put_forks(id)
            # thinking
            sleep(rng.random() / 100)

class PromptService(object):
    def __init__(self):
        self.semaphore = Semaphore(0)
        self.commandWindow = None
        self.response = None

    def setCommandWindow(self, window):
        self.commandWindow = window

    def requestInput(self, prompt):
        if self.commandWindow is None:
            raise RuntimeError("Command window hasn't registered itself")
        if prompt is None:
            prompt = ''

        self.commandWindow.prompt(prompt, 'standard-output', self.respond, 'standard-input')
        self.semaphore.acquire()

        if self.response is None:
            raise KeyboardInterrupt
        else:
            res = self.response
            self.response = None
            return str(res)

    def respond(self, value):
        self.response = value
        self.semaphore.release()

class Wc:
    def __init__(self):
        self.flush()

    def flush(self):
        self.sem = Semaphore(1)
        self.user = None
        self.waiting = []

    def used_by(self, who):
        self.user = who
        self.waiting.remove(who)

    def being_used_by(self, who):
        return self.user == who

    def acquire(self, a):
        return self.sem.acquire(a)

    def release(self):
        self.user = None
        self.sem.release()

    def enqueue(self, nick):
        self.waiting.append(nick)

    def is_waiting(self, nick):
        return (nick in self.waiting)
    def who(self):
        return self.user

class TestPubSubscribe(unittest.TestCase):
    def onMessage(self, message):
        self.assertTrue(len(message.payload_objects) > 0)
        msg_body = message.payload_objects[0].content
        self.assertIn(msg_body, MESSAGES)
        self.counter += 1
        if self.counter == len(MESSAGES):
            self.semaphore.release()

    def setUp(self):
        self.counter = 0
        self.semaphore = Semaphore(0)
        self.bw_client = Client()
        self.bw_client.setEntityFromFile(KEY_FILE)
        self.bw_client.overrideAutoChainTo(True)
        self.bw_client.subscribe(URI, self.onMessage)

    def tearDown(self):
        self.bw_client.close()

    def testPublishSubscribe(self):
        for msg in MESSAGES:
            po = PayloadObject((64, 0, 0, 0), None, msg)
            self.bw_client.publish(URI, payload_objects=(po,))
        self.semaphore.acquire()

    def test_output_handling(self):
        class Foo(ModuleBase):
            out = Output(channel_name="foo")
            inp = Input(channel_name="foo")
            clk = Clock()

            def __init__(self, cr, s):
                super().__init__(cr)
                self.s = s
                self.subscribe()

            def stop(self):
                self.clk.stop()

            def start(self):
                self.clk.frequency = 100
                self.clk.start()

            @activity(clk)
            async def tick(self):
                await self.out("hello, is it me you're looking for.")

            @activity(inp)
            async def qwerty(self, inp):
                assert (inp == "hello, is it me you're looking for.")
                self.s.release()

        cr = ChannelRegister()
        s = Semaphore(0)
        foo = Foo(cr, s)
        foo.start()
        self.assertTrue(s.acquire(timeout=0.1))
        foo.stop()

class Request_deque():
    from collections import deque
    def __init__(self, value=1):
        self.sema = Semaphore(value)
        self.time_stamp_q = deque()
        self.sync_lock = Lock()
    
    def acquire(self, blocking=True):
        if self.sema.acquire(blocking):
            # released under blocked mode or happened to have spare under
            #non-blocking mode
            return True, self.time_stamp_q.popleft()
        else:                    
            # non-blocking mode with unsuccessful acquiring
            return False, None
    
    def release(self, stop=False):
        with self.sync_lock:
            # need to guarantee the order matching between request and time
            #stamp, the operation shall be atomic. This could be rare to have
            #but unaffordable if any.
            if stop:
                self.time_stamp_q.append(None)
            else:
                self.time_stamp_q.append(dt.now())
            self.sema.release()

def wait_for_call(obj, target, callback=None):

    sem = Semaphore(0)
    result = WaitResult()

    unpatched = getattr(obj, target)

    def maybe_release(args, kwargs, res, exc_info):
        should_release = True
        if callable(callback):
            should_release = callback(args, kwargs, res, exc_info)

        if should_release:
            result.send(res, exc_info)
            sem.release()

    def wraps(*args, **kwargs):
        res = None
        exc_info = None
        try:
            res = unpatched(*args, **kwargs)
        except Exception:
            exc_info = sys.exc_info()

        maybe_release(args, kwargs, res, exc_info)

        if exc_info is not None:
            six.reraise(*exc_info)
        return res

    with patch.object(obj, target, new=wraps):
        yield result
        sem.acquire()

class BoundedHashSet(object):

    def __init__(self, capacity):
        """
        Lock is a mutex or a semaphore with count = 1
        This is used to guard the critical section and ensure mutual exclusion so only 1 thread
        has access at a time.

        Semaphore is to enforce capacity. Everytime sem.acquire() is called, capacity decrements
        by 1. When sem.release() is called, capacity increments by 1. If sem.acquire() is called
        when capacity  == 0, it blocks.
        :param capacity:
        :return:
        """
        self.mutex = Lock()
        self.st = set()
        self.sem = Semaphore(capacity)

    def add(self, item):
        if item not in self.st:
            self.sem.acquire()

        self.mutex.acquire()
        self.st.add(item)
        self.mutex.release()

    def erase(self, item):
        self.mutex.acquire()
        self.st.remove(item)
        self.mutex.release()
        self.sem.release()

class ThreadSemaphore(object):

    def __init__(self):

        self._semaphore = Semaphore(1)
        self._thread = None

    def acquire(self, wait=True):

        if self._thread is not currentThread():
            #print currentThread(), 'acquiring'
            result = self._semaphore.acquire(wait)
            if result:
                #print currentThread(), 'got it'
                self._thread = currentThread()

            return result

        return False

    def release(self):

        if self._thread is not currentThread():
            raise ValueError, 'current thread did not acquire semaphore'
        else:
            self._thread = None
            self._semaphore.release()

def call_on_main_thread(func, *args, **kwargs):
    done = Semaphore(0)
    # TODO use other name than "result"
    result = []

    def wrapped_call():
        try:
            res_cb = func(*args, **kwargs)
        except Exception, e:
            res_cb = Callback()
            res_cb(e)

        if not isinstance(res_cb, Callback):
            raise ValueError("Expected a monocle Callback from %r, got %r" % (func, res_cb))

        @_o
        def wait_for_result():
            try:
                res = yield res_cb
            except Exception, e:
                # TODO print traceback to a StringIO?
                res = e

            result.append(res)
            done.release()

class OneLaneBridge(object):
    """
    A one-lane bridge allows multiple cars to pass in either direction, but at any
    point in time, all cars on the bridge must be going in the same direction.

    Cars wishing to cross should call the cross function, once they have crossed
    they should call finished()
    """

    def __init__(self):
        self.direction = random.randrange(2)
        self.cars_on = Semaphore(0)
        self.num_cars_on = 0
        self.waiting = 0

    def cross(self,direction):
        """wait for permission to cross the bridge.  direction should be either
        north (0) or south (1)."""
        if (direction != self.direction) and self.num_cars_on:
            print "Waiting going direction %d" % direction
            self.waiting += 1
            self.cars_on.acquire()
        # Bridge is empty or == direction so we can take the opening and cross
        self.direction = direction
        self.num_cars_on += 1

    def finished(self, direction):
        self.num_cars_on -= 1
        if not self.num_cars_on:
            for car in range(self.waiting):
                self.cars_on.release()
            self.waiting = 0

 def _solve_mt(self, formula):
     solverThreads = []
     solution = None
     
     sem = Semaphore(0)
     
     for solver in self.solvers:
         sThread = _PortfolioThread(solver, formula, sem)
         solverThreads.append(sThread)
         
         sThread.start()
     
     # Wait for at least one thread to finish
     sem.acquire()
         
     for sThread in solverThreads:
         if solution is None and sThread.solution is not None:
             solution = sThread.solution
             if not self.benchMode:
                 self._benchmark = sThread.solver.getBenchmark()
         
         if not self.benchMode:
             sThread.solver.abort()
             
     for sThread in solverThreads:
         sThread.join()
         
     assert solution is not None, "Solver returned with invalid solution"
     
     if self.benchMode:
         self._benchmark = [sThread.solver.getBenchmark() for sThread in solverThreads]
         
     return solution