@Test
public void testPar() throws InterruptedException {
	for (int z = 0; z < 2; z++) {
		parPos = 0;
		final int parCount = 4;
		final Semaphore s = new Semaphore(parCount);
		final long start = System.currentTimeMillis();
		for (int i = 0; i < parCount; i++) {
			final Parser parser = new Parser(s);
			parser.start();
		}
		s.acquire(parCount);
		final long time = System.currentTimeMillis() - start;
		System.out.println("Finished parallel " + size + " in " + time + " => " + Math.round(size / time * 1000));
	}
}
 

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newCachedThreadPool();
    final Semaphore semaphore = new Semaphore(threadTotal);
    final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
    for (int i = 0; i < clientTotal; i++) {
        final int count = i;
        executorService.execute(() -> {
            try {
                semaphore.acquire();
                update(count);
                semaphore.release();
            } catch (Exception e) {
                log.error("exception", e);
            }
            countDownLatch.countDown();
        });
    }
    countDownLatch.await();
    executorService.shutdown();
    log.info("size:{}", list.size());
}
 

@Test
public void testSetPriorityOnNonexistentNode() throws InterruptedException {
  final DatabaseReference ref = IntegrationTestUtils.getRandomNode(masterApp);

  final Semaphore semaphore = new Semaphore(0);
  ref.setPriority(1, new DatabaseReference.CompletionListener() {
    @Override
    public void onComplete(DatabaseError error, DatabaseReference callbackRef) {
      assertEquals(ref, callbackRef);
      assertNotNull(error);
      semaphore.release(1);
    }
  });

  assertTrue(semaphore.tryAcquire(1, TestUtils.TEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
}
 

@Override
public void callService(final PluginParameters parameters, final InputStream in, final OutputStream out) throws IOException {
    final Graph graph = GraphManager.getDefault().getActiveGraph();

    // This is asynchronous, so we need a Semaphore.
    //
    final GraphNode graphNode = GraphNode.getGraphNode(graph);
    final VisualManager visualManager = graphNode.getVisualManager();
    final BufferedImage[] img1 = new BufferedImage[1];

    if (visualManager != null) {
        final Semaphore waiter = new Semaphore(0);
        visualManager.exportToBufferedImage(img1, waiter);
        waiter.acquireUninterruptibly();

        ImageIO.write(img1[0], "png", out);
    } else {
        throw new IOException("Graph image unavailable");
    }
}
 

/**
 * 初始化
 * 
 * @param threadCount
 * @param type
 */
private void init(int threadCount, Type type,Context context)
{
	// 获取我们应用的最大可用内存
	int maxMemory = (int) Runtime.getRuntime().maxMemory();
	int cacheMemory = maxMemory / 8;
	//注意此处要获取全局Context，避免引用Activity造成资源无法释放
	mContext=context.getApplicationContext();
	mLruCache = new LruCache<String, Bitmap>(cacheMemory){
		@Override
		protected int sizeOf(String key, Bitmap value)
		{
			//				return value.getAllocationByteCount();
			return value.getRowBytes() * value.getHeight(); //旧版本方法
		}

	};

	// 创建线程池
	mThreadPool = Executors.newFixedThreadPool(threadCount);
	mType = type;
	mSemaphoreThreadPool = new Semaphore(threadCount,true);
	mTaskQueue = new LinkedBlockingDeque<Runnable>();	
	initBackThread();
}
 

/** TODO */
public CalcSimilRunner(int iWordNGramSize_Min, int iWordNGramSize_Max, int iWord_Dmax,
        int iCharacterNGramSize_Min, int iCharacterNGramSize_Max, int iCharacter_Dmax,
        CategorizedFileEntry cfeCurEntry, List lCompareAgainst, Semaphore sSem, 
        boolean dDoCharNGrams, boolean dDoWordNGrams, PrintStream psOutStream,
        boolean bSilent, summaryEvaluator seCaller, int iWeightingMethod,
        boolean bProgress) {
    WordNGramSize_Min = iWordNGramSize_Min;
    WordNGramSize_Max = iWordNGramSize_Max;
    Word_Dmax = iWord_Dmax;
    CharacterNGramSize_Min = iCharacterNGramSize_Min;
    CharacterNGramSize_Max = iCharacterNGramSize_Max;
    Character_Dmax = iCharacter_Dmax;        
    CurEntry = cfeCurEntry;
    CompareAgainst = lCompareAgainst;
    Sem = sSem;
    DoCharNGrams = dDoCharNGrams;
    DoWordNGrams = dDoWordNGrams;
    OutStream = psOutStream;
    Silent = bSilent;
    Caller = seCaller;
    WeightingMethod = iWeightingMethod;
    Progress = bProgress;
}
 

public GenericCalcSimilRunner(int iNGramSize_Min, int iNGramSize_Max, int iDmax,
        CategorizedFileEntry cfeCurEntry, List lCompareAgainst, Semaphore sSem, 
        PrintStream psOutStream, boolean bSilent, summaryGenericEvaluator seCaller, 
        String sDocumentClass, String sComparatorClass,
        boolean bProgress) {
    NGramSize_Min = iNGramSize_Min;
    NGramSize_Max = iNGramSize_Max;
    Dmax = iDmax;        
    CurEntry = cfeCurEntry;
    CompareAgainst = lCompareAgainst;
    Sem = sSem;
    OutStream = psOutStream;
    Silent = bSilent;
    Caller = seCaller;
    Progress = bProgress;
    DocumentClass = sDocumentClass;
    ComparatorClass = sComparatorClass;
}
 

/**
 * @param consumer Runnable task.
 * @param grpId Group Id.
 * @param partId Partition Id.
 * @param exec Striped executor.
 */
public void stripedApplyPage(
    Consumer<PageMemoryEx> consumer,
    int grpId,
    int partId,
    StripedExecutor exec,
    Semaphore semaphore
) throws IgniteCheckedException {
    assert consumer != null;
    assert exec != null;
    assert semaphore != null;

    PageMemoryEx pageMem = getPageMemoryForCacheGroup(grpId);

    if (pageMem == null)
        return;

    stripedApply(() -> consumer.accept(pageMem), grpId, partId, exec, semaphore);
}
 

@Test
public void Success() throws Exception {
  final int N = 7;
  String[] urls = new String[N];
  Semaphore done = new Semaphore(0);
  Race.start(new SuccessProber(), urls, (int index) -> {
    assertTrue(index >= 0);
    assertTrue(index < N);
    done.release();
    if (done.availablePermits() > 1) {
      // Multiple success callbacks.
      fail();
    }
  });
  // Wait for listener to run.
  done.acquire();
}
 

public static void waitForEvents(DatabaseReference ref) {
  try {
    // Make sure queue is done and all events are queued
    IntegrationTestHelpers.waitForQueue(ref);
    // Next, all events were queued, make sure all events are done raised
    final Semaphore semaphore = new Semaphore(0);
    ref.getRepo()
        .postEvent(
            new Runnable() {
              @Override
              public void run() {
                semaphore.release();
              }
            });
    semaphore.acquire();
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}
 

/**
 * 
 * @param timeout
 * @param delayTimeMillis
 * @return 
 * @return boolean
 */
public boolean lock(int timeout,int delayTimeMillis){
    this.delayTimeMillis.set(delayTimeMillis);
    long startTime = System.currentTimeMillis();
    while (true) {
        try {
          //信号量为0，线程就会一直等待直到数据变成正数
            semaphore = new Semaphore(0);
            client.create(lockPath+"/lock", lockNodeData, CreateMode.EPHEMERAL);
            hasLock.set(true);
            return true;
        } catch (ZKNodeExistsException e) {
            try {
                 semaphore.acquire();
            } catch (InterruptedException interruptedException) {
                return false;
            }
        }
        //超时处理
        if (timeout > 0 && (System.currentTimeMillis() - startTime) >= timeout) {
            return false;
        }
    }
}
 

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newCachedThreadPool();
    final Semaphore semaphore = new Semaphore(threadTotal);
    final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
    for (int i = 0; i < clientTotal; i++) {
        final int count = i;
        executorService.execute(() -> {
            try {
                semaphore.acquire();
                update(count);
                semaphore.release();
            } catch (Exception e) {
                log.error("exception", e);
            }
            countDownLatch.countDown();
        });
    }
    countDownLatch.await();
    executorService.shutdown();
    log.info("size:{}", list.size());
}
 

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newCachedThreadPool();
    //Semaphore和CountDownLatch模拟并发
    final Semaphore semaphore = new Semaphore(threadTotal);
    final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
    for (int i = 0; i < clientTotal ; i++) {
        executorService.execute(() -> {
            try {
                semaphore.acquire();
                add();
                semaphore.release();
            } catch (Exception e) {
                log.error("exception", e);
            }
            countDownLatch.countDown();
        });
    }
    countDownLatch.await();
    executorService.shutdown();
    log.info("count:{}", count.get());
}
 

final int loop(int n) {
    final Semaphore sem = this.sem;
    int sum = 0;
    int x = 0;
    while (n-- > 0) {
        sem.acquireUninterruptibly();
        try {
            x = setValue(LoopHelpers.compute1(getValue()));
        }
        finally {
            sem.release();
        }
        sum += LoopHelpers.compute2(x);
    }
    return sum;
}
 

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newCachedThreadPool();
    final Semaphore semaphore = new Semaphore(threadTotal);
    final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
    for (int i = 0; i < clientTotal ; i++) {
        executorService.execute(() -> {
            try {
                semaphore.acquire();
                test();
                semaphore.release();
            } catch (Exception e) {
                log.error("exception", e);
            }
            countDownLatch.countDown();
        });
    }
    countDownLatch.await();
    executorService.shutdown();
    log.info("isHappened:{}", isHappened.get());
}
 

@Test
public void watchSingleFile() throws Exception {
    final File file = tempFolder.newFile();
    final Semaphore waitSemaphore = new Semaphore(0);
    FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, new FileWatchService.Listener() {
        @Override
        public void onChanged(String path) {
            assertThat(file.getAbsolutePath()).isEqualTo(path);
            waitSemaphore.release();
        }
    });
    fileWatchService.start();
    modifyFile(file);
    boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
    assertThat(result).isTrue();
}
 

@Test
public void multiThreadTest()
    throws Exception
{
    TreeMap<String, HashSet<String> > map=new DelayMap<String, HashSet<String>>();

    BandedUpdater<String, String> bu = new BandedUpdater<String,String>(map,16);
    Semaphore sem = new Semaphore(0);

    for(int i=0; i<100; i++)
    {
        new InsertThread(bu, sem).start();
    }
    sem.acquire(100);

    Assert.assertEquals(1, map.size());
    Assert.assertEquals(1000, map.get("a").size());

    
}
 

@Test
public void testUpdateSamePropertyTwice() throws Exception {
  String someNewValue = "someNewValue";

  Config otherConfig = ConfigService.getConfig(anotherNamespace);

  final Semaphore changes = new Semaphore(0);

  otherConfig.addChangeListener(new ConfigChangeListener() {
    @Override
    public void onChange(ConfigChangeEvent changeEvent) {
      changes.release();
    }
  });

  assertEquals("otherValue3", otherConfig.getProperty("key3", null));

  embeddedApollo.addOrModifyProperty(anotherNamespace, "key3", someNewValue);
  embeddedApollo.addOrModifyProperty(anotherNamespace, "key3", someNewValue);

  assertTrue(changes.tryAcquire(5, TimeUnit.SECONDS));
  assertEquals(someNewValue, otherConfig.getProperty("key3", null));
  assertEquals(0, changes.availablePermits());
}
 

private static void addChildrenAndWait(DatabaseReference reference, int amount)
    throws InterruptedException {
  addChildren(reference, amount, 0);
  final Semaphore semaphore = new Semaphore(0);
  reference
      .child("k-" + amount)
      .setValue(
          "last-value",
          new DatabaseReference.CompletionListener() {
            @Override
            public void onComplete(DatabaseError error, DatabaseReference ref) {
              semaphore.release();
            }
          });
  Assert.assertTrue(semaphore.tryAcquire(60, TimeUnit.SECONDS));
}
 

@Test
public void updateFiresCorrectEventWhenAllChildrenAreDeleted()
    throws DatabaseException, TestFailure, ExecutionException, TimeoutException,
        InterruptedException {
  List<DatabaseReference> refs = IntegrationTestHelpers.getRandomNode(2);
  DatabaseReference writer = refs.get(0);
  DatabaseReference reader = refs.get(1);

  ReadFuture writerFuture = ReadFuture.untilCountAfterNull(writer, 2);
  new WriteFuture(writer, new MapBuilder().put("a", 12).build()).timedGet();
  final Semaphore semaphore = new Semaphore(0);
  ReadFuture readerFuture =
      new ReadFuture(
          reader,
          new ReadFuture.CompletionCondition() {
            @Override
            public boolean isComplete(List<EventRecord> events) {
              if (events.size() == 1) {
                semaphore.release();
              }
              return events.size() == 2;
            }
          });

  IntegrationTestHelpers.waitFor(semaphore);

  writer.updateChildren(new MapBuilder().put("a", null).build());
  DataSnapshot snap = writerFuture.timedGet().get(1).getSnapshot();

  assertNull(snap.getValue());

  snap = readerFuture.timedGet().get(1).getSnapshot();

  assertNull(snap.getValue());
}
 

public void start(int queueSize) {
	// THREAD = 1
	//
	// The parallel processing is within the de-compressor itself - here we
	// want
	// to ensure a certain order of processing - if need more than one
	// thread then
	// have a look how the de-compressed data are sent to the GUI (!)

	executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

	executor.setThreadFactory(new DefaultThreadFactoryEx("DeCompressorEngine"));

	// Rejection Policy
	//
	// Blocking pattern when queue full; that means we're not decompressing
	// fast enough; when our queue is full
	// then the network receiving thread is going to stop reading from the
	// assisted side which in turn is going
	// to slow down sending its capture leaving us some time to catch up.
	//
	// Having our queue full is quite unlikely; I would say the network will
	// limit the number of capture/tiles
	// being sent and I guess that decompressing is much faster then
	// compressing (unless our PC is quite weak
	// compared to the assisted one; let's not forget the JAVA capture is
	// awful regarding the performance as
	// well => should be fine here.

	semaphore = new Semaphore(queueSize, true);
}
 

@SuppressWarnings("unchecked")
private AttributeSubscriptionListener<Integer> prepareOnErrorListenerMock(final Semaphore onErrorSemaphore,
                                                                          JoynrRuntimeException expectation) {
    AttributeSubscriptionListener<Integer> integerListener = mock(AttributeSubscriptionListener.class);

    doAnswer(new Answer<Object>() {
        @Override
        public Object answer(InvocationOnMock invocation) {
            onErrorSemaphore.release();
            return null;
        }
    }).when(integerListener).onError(expectation);
    return integerListener;
}
 

public void testTryAcquire_timeout(boolean fair) {
    Semaphore s = new Semaphore(0, fair);
    long startTime = System.nanoTime();
    try { assertFalse(s.tryAcquire(timeoutMillis(), MILLISECONDS)); }
    catch (InterruptedException e) { threadUnexpectedException(e); }
    assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
}
 

@Test
public void testExecuteWritesStderrToTheStderrListener() throws Throwable {
    final JobExecutor jobExecutor = getInstance();
    final String msgSuppliedToEcho = generateRandomString();
    final String bashArg = "echo " + msgSuppliedToEcho + " 1>&2"; // TODO: Naughty.
    final PersistedJob req =
            standardRequestWithCommand("bash", "-c", bashArg);
    final AtomicReference<byte[]> bytesEchoedToStderr = new AtomicReference<>(new byte[]{});
    final Subject<byte[]> stderrSubject = PublishSubject.create();

    stderrSubject.subscribe(bytes ->
            bytesEchoedToStderr.getAndUpdate(existingBytes ->
                    Bytes.concat(existingBytes, bytes)));

    final Semaphore s = new Semaphore(1);
    s.acquire();
    stderrSubject.doOnComplete(s::release).subscribe();

    final JobEventListeners listeners =
            createStderrListener(stderrSubject);

    jobExecutor.execute(req, listeners);

    s.tryAcquire(TestConstants.DEFAULT_TIMEOUT, MILLISECONDS);

    final String stringFromStderr = new String(bytesEchoedToStderr.get()).trim();
    assertThat(stringFromStderr).isEqualTo(msgSuppliedToEcho);
}
 

@Test
public void testDataUpdate() throws InterruptedException {
  final Repo repo = newRepo();
  final List<DataSnapshot> events = new ArrayList<>();
  QuerySpec spec = new QuerySpec(new Path("/foo"), QueryParams.DEFAULT_PARAMS);
  addCallback(
      repo, new ValueEventRegistration(repo, newValueEventListener(events), spec));

  final Semaphore semaphore = new Semaphore(0);
  repo.scheduleNow(new Runnable() {
    @Override
    public void run() {
      repo.onDataUpdate(ImmutableList.of("foo"), "testData", false, null);
      semaphore.release();
    }
  });
  waitFor(semaphore);
  assertEquals(1, events.size());
  assertNotNull(events.get(0));
  assertEquals("testData", events.get(0).getValue(String.class));

  final Map<String, String> update = ImmutableMap.of("key1", "value1");
  repo.scheduleNow(new Runnable() {
    @Override
    public void run() {
      repo.onDataUpdate(ImmutableList.of("foo"), update, true, null);
      semaphore.release();
    }
  });
  waitFor(semaphore);
  assertEquals(2, events.size());
  assertNotNull(events.get(1));
  assertEquals(update, events.get(1).getValue());
}
 

/** {@inheritDoc} */
@Override
public void writeRaw(IoBuffer out) {
    if (ioSession != null) {
        final Semaphore lock = getLock();
        while (!isClosed()) {
            boolean acquired = false;
            try {
                acquired = lock.tryAcquire(10, TimeUnit.MILLISECONDS);
                if (acquired) {
                    if (log.isTraceEnabled()) {
                        log.trace("Writing raw message");
                    }
                    ioSession.write(out);
                    break;
                }
            } catch (InterruptedException e) {
                log.warn("Interrupted while waiting for write lock (writeRaw). State: {}", RTMP.states[state.getState()], e);
                String exMsg = e.getMessage();
                // if the exception cause is null break out of here to prevent looping until closed
                if (exMsg == null || exMsg.indexOf("null") >= 0) {
                    log.debug("Exception writing to connection: {}", this);
                    break;
                }
            } finally {
                if (acquired) {
                    lock.release();
                }
            }
        }
    }
}
 

public static void runNow(Runnable r) {
    if (Platform.isFxApplicationThread()) {
        r.run();
    } else {
        Semaphore s = new Semaphore(0);
        Platform.runLater(()->{
            r.run();
            synchronized (s) {
                s.release();
            }
        });
        s.acquireUninterruptibly();
    }
}
 

private void init(int threadCount, Type type) {
    mPoolThread = new Thread() {
        @Override
        public void run() {
            Looper.prepare();
            mPoolThreadHandler = new Handler() {
                @Override
                public void handleMessage(Message msg) {
                    mThreadPool.execute(getTask());
                    try {
                        mSemaphoreThreadPool.acquire();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            mSemaphorePoolThreadHandler.release();
            Looper.loop();
        }
    };
    mPoolThread.start();
    
    int maxMemory = (int) Runtime.getRuntime().maxMemory();
    int cacheMemory = maxMemory / 8;
    mLruCache = new LruCache<String, Bitmap>(cacheMemory) {
        @Override
        protected int sizeOf(String key, Bitmap value) {
            return value.getRowBytes() * value.getHeight();
        }
    };

    mThreadPool = Executors.newFixedThreadPool(threadCount);
    mTaskQueue = new LinkedList<Runnable>();
    mType = type;
    mSemaphoreThreadPool = new Semaphore(threadCount);

}
 

@Override
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws ProtocolCodecException {
    // get the connection from the session
    String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
    log.trace("Session id: {}", sessionId);
    RTMPConnection conn = (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
    if (conn != null) {
        Red5.setConnectionLocal(conn);
        final Semaphore lock = conn.getEncoderLock();
        try {
            // acquire the encoder lock
            lock.acquire();
            // get the buffer
            final IoBuffer buf = message instanceof IoBuffer ? (IoBuffer) message : getEncoder().encode(message);
            if (buf != null) {
                if (log.isTraceEnabled()) {
                    log.trace("Writing output data: {}", Hex.encodeHexString(buf.array()));
                }
                out.write(buf);
            } else {
                log.trace("Response buffer was null after encoding");
            }
        } catch (Exception ex) {
            log.error("Exception during encode", ex);
        } finally {
            lock.release();
            Red5.setConnectionLocal(null);
        }
    } else {
        log.debug("Connection is no longer available for encoding, may have been closed already");
    }
}
 

/**
 * 対応しているGL|ESのバージョンを取得
 * XXX GLES30はAPI>=18以降なんだけどAPI=18でもGLコンテキスト生成に失敗する端末があるのでAP1>=21に変更
 * API>=21でGL_OES_EGL_image_external_essl3に対応していれば3, そうでなければ2を返す
 * @return
 */
public static int getSupportedGLVersion() {
	if (sSupportedGLVersion < 1) {
		// 一度も実行されていない時
		final AtomicInteger result = new AtomicInteger(1);
		final Semaphore sync = new Semaphore(0);
		final GLContext context = new GLContext(3, null, 0);
		// ダミースレッド上でEGL/GLコンテキストを生成してエクステンション文字列をチェックする
		new Thread(new Runnable() {
			@Override
			public void run() {
				context.initialize();
				String extensions = GLES20.glGetString(GLES20.GL_EXTENSIONS); // API >= 8
				if (DEBUG) Log.i(TAG, "getSupportedGLVersion:" + extensions);
				if ((extensions == null) || !extensions.contains("GL_OES_EGL_image_external")) {
					result.set(1);
				} else if ((Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP) && context.isGLES3()) {
					extensions = GLES30.glGetString(GLES30.GL_EXTENSIONS); 	// API >= 18
					result.set((extensions != null) && extensions.contains("GL_OES_EGL_image_external_essl3")
						? 3 : 2);
				} else {
					result.set(2);
				}
				context.release();
				sync.release();
			}
		}).start();
		try {
			sync.tryAcquire(500, TimeUnit.MILLISECONDS);
			sSupportedGLVersion = result.get();
		} catch (final InterruptedException e) {
			// ignore
		}
	}
	if (DEBUG) Log.i(TAG, "getSupportedGLVersion:" + sSupportedGLVersion);
	return sSupportedGLVersion;
}