int doIO(ByteBuffer buf, int ops) throws IOException {
    
    /* For now only one thread is allowed. If user want to read or write
     * from multiple threads, multiple streams could be created. In that
     * case multiple threads work as well as underlying channel supports it.
     */
    if (!buf.hasRemaining()) {
      throw new IllegalArgumentException("Buffer has no data left.");
      //or should we just return 0?
    }

    while (buf.hasRemaining()) {
      if (closed) {
        return -1;
      }

      try {
        int n = performIO(buf);
        if (n != 0) {
          // successful io or an error.
          return n;
        }
      } catch (IOException e) {
        if (!channel.isOpen()) {
          closed = true;
        }
        throw e;
      }

      //now wait for socket to be ready.
      int count = 0;
      try {
        count = SelectorPool.select(channel, ops, timeout);
      } catch (IOException e) { //unexpected IOException.
        closed = true;
        throw e;
      } 

      if (count == 0) {
        throw new SocketTimeoutException(timeoutExceptionString(channel,
                                                                timeout, ops));
      }
      // otherwise the socket should be ready for io.
    }
    
    return 0; // does not reach here.
  }
@Test
  public void testSocketIOWithTimeoutByMultiThread() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);
    Runnable ioTask = () -> {
      try {
        Pipe pipe = Pipe.open();
        try (Pipe.SourceChannel source = pipe.source();
             InputStream in = new SocketInputStream(source, TIMEOUT);
             Pipe.SinkChannel sink = pipe.sink();
             OutputStream out = new SocketOutputStream(sink, TIMEOUT)) {

          byte[] writeBytes = TEST_STRING.getBytes();
          byte[] readBytes = new byte[writeBytes.length];
          latch.await();

          out.write(writeBytes);
          doIO(null, out, TIMEOUT);

          in.read(readBytes);
          assertArrayEquals(writeBytes, readBytes);
          doIO(in, null, TIMEOUT);
        }
      } catch (Exception e) {
        fail(e.getMessage());
      }
    };

    int threadCnt = 64;
    ExecutorService threadPool = Executors.newFixedThreadPool(threadCnt);
    for (int i = 0; i < threadCnt; ++i) {
      threadPool.submit(ioTask);
    }

    Thread.sleep(1000);
    latch.countDown();

    threadPool.shutdown();
    assertTrue(threadPool.awaitTermination(3, TimeUnit.SECONDS));
  }