/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package tachyon.client.concurrent;

import java.io.IOException;
import java.util.List;

import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;

import com.google.common.collect.Lists;

import tachyon.Constants;
import tachyon.LocalTachyonClusterResource;
import tachyon.client.ClientContext;
import tachyon.client.StreamOptionUtils;
import tachyon.client.TachyonFSTestUtils;
import tachyon.client.file.FileInStream;
import tachyon.client.file.TachyonFile;
import tachyon.client.file.TachyonFileSystem;
import tachyon.client.file.options.OutStreamOptions;
import tachyon.conf.TachyonConf;
import tachyon.exception.FileDoesNotExistException;
import tachyon.exception.TachyonException;
import tachyon.util.io.PathUtils;

/**
 * Tests the concurrency of {@link FileInStream}.
 */
public final class FileInStreamConcurrencyIntegrationTest {
  private static final int BLOCK_SIZE = 30;

  @ClassRule
  public static LocalTachyonClusterResource sLocalTachyonClusterResource =
      new LocalTachyonClusterResource(Constants.GB, Constants.KB, BLOCK_SIZE);
  private static TachyonFileSystem sTfs = null;
  private static TachyonConf sTachyonConf;
  private static OutStreamOptions sWriteTachyon;

  @BeforeClass
  public static final void beforeClass() throws Exception {
    sTfs = sLocalTachyonClusterResource.get().getClient();
    sTachyonConf = sLocalTachyonClusterResource.get().getMasterTachyonConf();
    sWriteTachyon = StreamOptionUtils.getOutStreamOptionsWriteTachyon(sTachyonConf);
  }

  /**
   * Tests the concurrent read of {@link FileInStream}.
   */
  @Test
  public void FileInStreamConcurrencyTest() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    TachyonFile f =
        TachyonFSTestUtils.createByteFile(sTfs, uniqPath, BLOCK_SIZE * 2, sWriteTachyon);

    List<Thread> threads = Lists.newArrayList();
    for (int i = 0; i < ClientContext.getConf().getInt(Constants.USER_BLOCK_MASTER_CLIENT_THREADS)
        * 10; i ++) {
      threads.add(new Thread(new FileRead(f)));
    }

    ConcurrencyTestUtils.assertConcurrent(threads, 100);
  }

  class FileRead implements Runnable {
    private final TachyonFile mFile;

    FileRead(TachyonFile file) {
      mFile = file;
    }

    @Override
    public void run() {
      try {
        FileInStream stream = sTfs.getInStream(mFile);
        stream.read();
        stream.close();
      } catch (IOException e) {
        throw new RuntimeException(e);
      } catch (FileDoesNotExistException e) {
        throw new RuntimeException(e);
      } catch (TachyonException e) {
        throw new RuntimeException(e);
      }
    }
  }
}

@@ -826,36 +826,8 @@
 t;%0A%0A
-import java.io.IOException;%0A
 impo
@@ -841,24 +841,24 @@
 .util.List;%0A
+
 %0Aimport org.
@@ -924,24 +924,66 @@
 unit.Test;%0A%0A
+import com.google.common.base.Throwables;%0A
 import com.g
@@ -1418,103 +1418,8 @@
 nf;%0A
-import tachyon.exception.FileDoesNotExistException;%0Aimport tachyon.exception.TachyonException;%0A
 impo
@@ -1610,16 +1610,149 @@
 ZE = 30;
+%0A  private static final int READ_THREADS_NUM =%0A      ClientContext.getConf().getInt(Constants.USER_BLOCK_MASTER_CLIENT_THREADS) * 10;
 %0A%0A  @Cla
@@ -2726,95 +2726,24 @@
 i %3C 
-ClientContext.getConf().getInt(Constants.USER_BLOCK_MASTER_CLIENT_THREADS)%0A        * 10
+READ_THREADS_NUM
 ; i 
@@ -3169,18 +3169,16 @@
  catch (
-IO
 Exceptio
@@ -3196,195 +3196,28 @@
     
-t
+T
 hrow
- new RuntimeException(e);%0A      %7D catch (FileDoesNotExistException e) %7B%0A        throw new RuntimeException(e);%0A      %7D catch (TachyonException e) %7B%0A        throw new RuntimeException
+ables.propagate
 (e);

address aaudiber's comments