{"action":"created","comment":{"url":"https://api.github.com/repos/Alluxio/alluxio/pulls/comments/63085212","id":63085212,"diff_hunk":"@@ -416,78 +351,116 @@ private long getCurrentBlockId() {\n   }\n \n   /**\n-   * @param pos the position to get the block size for\n-   * @return the size of the current block\n-   */\n-  protected long getBlockSize(long pos) {\n-    // The size of the last block, 0 if it is equal to the normal block size\n-    long lastBlockSize = mFileLength % mBlockSize;\n-    // If we are not in the last block or if the last block is equal to the normal block size,\n-    // return the normal block size. Otherwise return the block size of the last block.\n-    if (mFileLength - pos > lastBlockSize) {\n-      return mBlockSize;\n-    } else {\n-      return lastBlockSize;\n-    }\n-  }\n-\n-  /**\n    * Logs IO exceptions thrown in response to the worker cache request. If the exception is not an\n    * expected exception, a warning will be logged with the stack trace.\n    */\n-  private void logCacheStreamIOException(IOException e) {\n+  private void handleCacheStreamIOException(IOException e) throws IOException {\n     if (e.getCause() instanceof BlockAlreadyExistsException) {\n       LOG.warn(BLOCK_ID_EXISTS_SO_NOT_CACHED, getCurrentBlockId());\n     } else {\n       LOG.warn(BLOCK_ID_NOT_CACHED, getCurrentBlockId(), e);\n     }\n+    closeOrCancelCacheStream();\n   }\n \n   /**\n-   * Similar to {@link #checkAndAdvanceBlockInStream()}, but a specific position can be specified\n-   * and the stream pointer will be at that offset after this method completes.\n+   * Only updates {@link #mCurrentCacheStream}, {@link #mCurrentBlockInStream} and\n+   * {@link #mStreamBlockId} to be in-sync the current block (i.e. {@link #getCurrentBlockId()}).\n+   * This function can be called multiple times without side effect. It is recommended to be\n+   * invoked before every read and seek.\n    *\n-   * @param newPos the new position to set the stream to\n-   * @throws IOException if the stream at the specified position cannot be opened\n+   * @throws IOException if the next cache stream or block stream cannot be created\n    */\n-  private void seekBlockInStream(long newPos) throws IOException {\n-    long oldBlockId = getCurrentBlockId();\n-    mPos = newPos;\n-    closeCacheStream();\n+  private void updateStreams() throws IOException {\n     long currentBlockId = getCurrentBlockId();\n-\n-    if (oldBlockId != currentBlockId) {\n+    if (shouldUpdateStreams(currentBlockId)) {\n+      // The following two function handle negative currentBlockId (i.e. the end of file)\n+      // correctly.\n       updateBlockInStream(currentBlockId);\n-      // Reading next block entirely.\n-      if (mPos % mBlockSize == 0 && mShouldCacheCurrentBlock) {\n-        try {\n-          WorkerNetAddress address = mLocationPolicy.getWorkerForNextBlock(\n-              mContext.getAlluxioBlockStore().getWorkerInfoList(), getBlockSizeAllocation(mPos));\n-          mCurrentCacheStream = createCacheStream(currentBlockId, getBlockSize(mPos), address);\n-        } catch (IOException e) {\n-          logCacheStreamIOException(e);\n-          mShouldCacheCurrentBlock = false;\n-        } catch (AlluxioException e) {\n-          LOG.warn(BLOCK_ID_NOT_CACHED, currentBlockId, e);\n-          mShouldCacheCurrentBlock = false;\n+      updateCacheStream(currentBlockId);\n+      mStreamBlockId = currentBlockId;\n+    }\n+  }\n+\n+  /**\n+   * Update {@link #mCurrentCacheStream}. The following preconditions are checked inside:\n+   *   1. {@link #mCurrentCacheStream} is either done or null.\n+   *   2. EOF is reached or {@link #mCurrentBlockInStream} must be valid.\n+   * After this call, {@link #mCurrentCacheStream} is either null or freshly created.\n+   * {@link #mCurrentCacheStream} is created only if the block is not cached in a chosen machine\n+   * and mPos is at the beginning of a block.\n+   * This function is only called by {@link #updateStreams()}.\n+   *\n+   * @param blockId cached result of {@link #getCurrentBlockId()}\n+   * @throws IOException if the next cache stream cannot be created\n+   */\n+  private void updateCacheStream(long blockId) throws IOException {\n+    // We should really only close a cache stream here. This check is to verify this.\n+    Preconditions.checkState(mCurrentCacheStream == null || mCurrentCacheStream.remaining() == 0);\n+    closeOrCancelCacheStream();\n+    Preconditions.checkState(mCurrentCacheStream == null);\n+\n+    if (blockId < 0) {\n+      // End of file.\n+      return;\n+    }\n+    Preconditions.checkNotNull(mCurrentBlockInStream);\n+    if (!mShouldCache || mCurrentBlockInStream instanceof LocalBlockInStream) {\n+      return;\n+    }\n+\n+    // Unlike updateBlockInStream below, we never start a block cache stream if mPos is in the\n+    // middle of a block.\n+    if (mPos % mBlockSize != 0) {\n+      return;\n+    }\n+\n+    try {\n+      WorkerNetAddress address = mLocationPolicy.getWorkerForNextBlock(\n+          mContext.getAlluxioBlockStore().getWorkerInfoList(), getBlockSizeAllocation(mPos));\n+      // Don't cache the block to somewhere that already has it.\n+      // TODO(andrew,peis): Filter the workers provided to the location policy to not include\n+      // workers which already contain the block. See ALLUXIO-1816.\n+      if (mCurrentBlockInStream instanceof RemoteBlockInStream) {\n+        WorkerNetAddress readAddress =\n+            ((RemoteBlockInStream) mCurrentBlockInStream).getWorkerNetAddress();\n+        // Try to avoid an RPC.\n+        if (readAddress.equals(address)) {\n+          return;\n+        }\n+        BlockInfo blockInfo = mContext.getAlluxioBlockStore().getInfo(blockId);\n+        for (BlockLocation location : blockInfo.getLocations()) {\n+          if (address.equals(location.getWorkerAddress())) {\n+            return;\n+          }\n         }\n-      } else {\n-        mShouldCacheCurrentBlock = false;\n       }\n+      // If we reach here, we need to cache.\n+      mCurrentCacheStream =\n+          mContext.getAlluxioBlockStore().getOutStream(blockId, getBlockSize(mPos), address);\n+    } catch (IOException e) {\n+      handleCacheStreamIOException(e);\n+    } catch (AlluxioException e) {\n+      LOG.warn(BLOCK_ID_NOT_CACHED, blockId, e);\n     }\n   }\n \n   /**\n-   * Helper method to {@link #checkAndAdvanceBlockInStream()} and {@link #seekBlockInStream(long)}.\n-   * The current {@link BlockInStream} will be closed and a new {@link BlockInStream} for the given\n-   * blockId will be opened at position 0.\n+   * Update {@link #mCurrentBlockInStream} to be in-sync with mPos's block. The new block\n+   * stream created with be at position 0.\n+   * This function is only called in {@link #updateStreams()}.\n    *\n-   * @param blockId blockId to set the {@link #mCurrentBlockInStream} to read\n+   * @param blockId cached result of {@link #getCurrentBlockId()}\n    * @throws IOException if the next {@link BlockInStream} cannot be obtained\n    */\n   private void updateBlockInStream(long blockId) throws IOException {\n     if (mCurrentBlockInStream != null) {\n       mCurrentBlockInStream.close();\n+      mCurrentBlockInStream = null;\n+    }\n+    if (blockId < 0) {","path":"core/client/src/main/java/alluxio/client/file/FileInStream.java","position":523,"original_position":524,"commit_id":"ffa6381a9fed787b7e2b0933586cdd9e2f33597d","original_commit_id":"773ca7361887db411f79363136340e689c6c7039","user":{"login":"apc999","id":1413748,"avatar_url":"https://avatars.githubusercontent.com/u/1413748?v=3","gravatar_id":"","url":"https://api.github.com/users/apc999","html_url":"https://github.com/apc999","followers_url":"https://api.github.com/users/apc999/followers","following_url":"https://api.github.com/users/apc999/following{/other_user}","gists_url":"https://api.github.com/users/apc999/gists{/gist_id}","starred_url":"https://api.github.com/users/apc999/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apc999/subscriptions","organizations_url":"https://api.github.com/users/apc999/orgs","repos_url":"https://api.github.com/users/apc999/repos","events_url":"https://api.github.com/users/apc999/events{/privacy}","received_events_url":"https://api.github.com/users/apc999/received_events","type":"User","site_admin":false},"body":"could you comment this in the code?","created_at":"2016-05-12T19:50:33Z","updated_at":"2016-05-12T19:50:33Z","html_url":"https://github.com/Alluxio/alluxio/pull/3208#discussion_r63085212","pull_request_url":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208","_links":{"self":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/comments/63085212"},"html":{"href":"https://github.com/Alluxio/alluxio/pull/3208#discussion_r63085212"},"pull_request":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208"}}},"pull_request":{"url":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208","id":69456834,"html_url":"https://github.com/Alluxio/alluxio/pull/3208","diff_url":"https://github.com/Alluxio/alluxio/pull/3208.diff","patch_url":"https://github.com/Alluxio/alluxio/pull/3208.patch","issue_url":"https://api.github.com/repos/Alluxio/alluxio/issues/3208","number":3208,"state":"open","locked":false,"title":"[ALLUXIO-1942] Refactor FileInStream","user":{"login":"peisun1115","id":18294241,"avatar_url":"https://avatars.githubusercontent.com/u/18294241?v=3","gravatar_id":"","url":"https://api.github.com/users/peisun1115","html_url":"https://github.com/peisun1115","followers_url":"https://api.github.com/users/peisun1115/followers","following_url":"https://api.github.com/users/peisun1115/following{/other_user}","gists_url":"https://api.github.com/users/peisun1115/gists{/gist_id}","starred_url":"https://api.github.com/users/peisun1115/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/peisun1115/subscriptions","organizations_url":"https://api.github.com/users/peisun1115/orgs","repos_url":"https://api.github.com/users/peisun1115/repos","events_url":"https://api.github.com/users/peisun1115/events{/privacy}","received_events_url":"https://api.github.com/users/peisun1115/received_events","type":"User","site_admin":false},"body":"Preparation PR to fix Parquet related bugs.\r\n\r\nThis PR also fixes some small bugs and simplifies logics in FileInStream and UnknownLengthFileInStream","created_at":"2016-05-10T06:54:27Z","updated_at":"2016-05-12T19:50:33Z","closed_at":null,"merged_at":null,"merge_commit_sha":"90a03a44611e4fdbb399d6107489cda874f3b6e8","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208/commits","review_comments_url":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208/comments","review_comment_url":"https://api.github.com/repos/Alluxio/alluxio/pulls/comments{/number}","comments_url":"https://api.github.com/repos/Alluxio/alluxio/issues/3208/comments","statuses_url":"https://api.github.com/repos/Alluxio/alluxio/statuses/ffa6381a9fed787b7e2b0933586cdd9e2f33597d","head":{"label":"peisun1115:ParquetFix","ref":"ParquetFix","sha":"ffa6381a9fed787b7e2b0933586cdd9e2f33597d","user":{"login":"peisun1115","id":18294241,"avatar_url":"https://avatars.githubusercontent.com/u/18294241?v=3","gravatar_id":"","url":"https://api.github.com/users/peisun1115","html_url":"https://github.com/peisun1115","followers_url":"https://api.github.com/users/peisun1115/followers","following_url":"https://api.github.com/users/peisun1115/following{/other_user}","gists_url":"https://api.github.com/users/peisun1115/gists{/gist_id}","starred_url":"https://api.github.com/users/peisun1115/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/peisun1115/subscriptions","organizations_url":"https://api.github.com/users/peisun1115/orgs","repos_url":"https://api.github.com/users/peisun1115/repos","events_url":"https://api.github.com/users/peisun1115/events{/privacy}","received_events_url":"https://api.github.com/users/peisun1115/received_events","type":"User","site_admin":false},"repo":{"id":55541423,"name":"alluxio","full_name":"peisun1115/alluxio","owner":{"login":"peisun1115","id":18294241,"avatar_url":"https://avatars.githubusercontent.com/u/18294241?v=3","gravatar_id":"","url":"https://api.github.com/users/peisun1115","html_url":"https://github.com/peisun1115","followers_url":"https://api.github.com/users/peisun1115/followers","following_url":"https://api.github.com/users/peisun1115/following{/other_user}","gists_url":"https://api.github.com/users/peisun1115/gists{/gist_id}","starred_url":"https://api.github.com/users/peisun1115/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/peisun1115/subscriptions","organizations_url":"https://api.github.com/users/peisun1115/orgs","repos_url":"https://api.github.com/users/peisun1115/repos","events_url":"https://api.github.com/users/peisun1115/events{/privacy}","received_events_url":"https://api.github.com/users/peisun1115/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/peisun1115/alluxio","description":"Memory-Speed Virtual Distributed Storage System","fork":true,"url":"https://api.github.com/repos/peisun1115/alluxio","forks_url":"https://api.github.com/repos/peisun1115/alluxio/forks","keys_url":"https://api.github.com/repos/peisun1115/alluxio/keys{/key_id}","collaborators_url":"https://api.github.com/repos/peisun1115/alluxio/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/peisun1115/alluxio/teams","hooks_url":"https://api.github.com/repos/peisun1115/alluxio/hooks","issue_events_url":"https://api.github.com/repos/peisun1115/alluxio/issues/events{/number}","events_url":"https://api.github.com/repos/peisun1115/alluxio/events","assignees_url":"https://api.github.com/repos/peisun1115/alluxio/assignees{/user}","branches_url":"https://api.github.com/repos/peisun1115/alluxio/branches{/branch}","tags_url":"https://api.github.com/repos/peisun1115/alluxio/tags","blobs_url":"https://api.github.com/repos/peisun1115/alluxio/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/peisun1115/alluxio/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/peisun1115/alluxio/git/refs{/sha}","trees_url":"https://api.github.com/repos/peisun1115/alluxio/git/trees{/sha}","statuses_url":"https://api.github.com/repos/peisun1115/alluxio/statuses/{sha}","languages_url":"https://api.github.com/repos/peisun1115/alluxio/languages","stargazers_url":"https://api.github.com/repos/peisun1115/alluxio/stargazers","contributors_url":"https://api.github.com/repos/peisun1115/alluxio/contributors","subscribers_url":"https://api.github.com/repos/peisun1115/alluxio/subscribers","subscription_url":"https://api.github.com/repos/peisun1115/alluxio/subscription","commits_url":"https://api.github.com/repos/peisun1115/alluxio/commits{/sha}","git_commits_url":"https://api.github.com/repos/peisun1115/alluxio/git/commits{/sha}","comments_url":"https://api.github.com/repos/peisun1115/alluxio/comments{/number}","issue_comment_url":"https://api.github.com/repos/peisun1115/alluxio/issues/comments{/number}","contents_url":"https://api.github.com/repos/peisun1115/alluxio/contents/{+path}","compare_url":"https://api.github.com/repos/peisun1115/alluxio/compare/{base}...{head}","merges_url":"https://api.github.com/repos/peisun1115/alluxio/merges","archive_url":"https://api.github.com/repos/peisun1115/alluxio/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/peisun1115/alluxio/downloads","issues_url":"https://api.github.com/repos/peisun1115/alluxio/issues{/number}","pulls_url":"https://api.github.com/repos/peisun1115/alluxio/pulls{/number}","milestones_url":"https://api.github.com/repos/peisun1115/alluxio/milestones{/number}","notifications_url":"https://api.github.com/repos/peisun1115/alluxio/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/peisun1115/alluxio/labels{/name}","releases_url":"https://api.github.com/repos/peisun1115/alluxio/releases{/id}","deployments_url":"https://api.github.com/repos/peisun1115/alluxio/deployments","created_at":"2016-04-05T20:01:18Z","updated_at":"2016-04-05T20:01:22Z","pushed_at":"2016-05-12T18:45:25Z","git_url":"git://github.com/peisun1115/alluxio.git","ssh_url":"git@github.com:peisun1115/alluxio.git","clone_url":"https://github.com/peisun1115/alluxio.git","svn_url":"https://github.com/peisun1115/alluxio","homepage":"http://www.alluxio.org/","size":49484,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":0,"forks":1,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"Alluxio:master","ref":"master","sha":"769a82f30c36c1248f515fe57a211c65fe29d700","user":{"login":"Alluxio","id":16203694,"avatar_url":"https://avatars.githubusercontent.com/u/16203694?v=3","gravatar_id":"","url":"https://api.github.com/users/Alluxio","html_url":"https://github.com/Alluxio","followers_url":"https://api.github.com/users/Alluxio/followers","following_url":"https://api.github.com/users/Alluxio/following{/other_user}","gists_url":"https://api.github.com/users/Alluxio/gists{/gist_id}","starred_url":"https://api.github.com/users/Alluxio/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Alluxio/subscriptions","organizations_url":"https://api.github.com/users/Alluxio/orgs","repos_url":"https://api.github.com/users/Alluxio/repos","events_url":"https://api.github.com/users/Alluxio/events{/privacy}","received_events_url":"https://api.github.com/users/Alluxio/received_events","type":"Organization","site_admin":false},"repo":{"id":7276954,"name":"alluxio","full_name":"Alluxio/alluxio","owner":{"login":"Alluxio","id":16203694,"avatar_url":"https://avatars.githubusercontent.com/u/16203694?v=3","gravatar_id":"","url":"https://api.github.com/users/Alluxio","html_url":"https://github.com/Alluxio","followers_url":"https://api.github.com/users/Alluxio/followers","following_url":"https://api.github.com/users/Alluxio/following{/other_user}","gists_url":"https://api.github.com/users/Alluxio/gists{/gist_id}","starred_url":"https://api.github.com/users/Alluxio/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Alluxio/subscriptions","organizations_url":"https://api.github.com/users/Alluxio/orgs","repos_url":"https://api.github.com/users/Alluxio/repos","events_url":"https://api.github.com/users/Alluxio/events{/privacy}","received_events_url":"https://api.github.com/users/Alluxio/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/Alluxio/alluxio","description":"Alluxio, formerly Tachyon, Memory Speed Virtual Distributed Storage System","fork":false,"url":"https://api.github.com/repos/Alluxio/alluxio","forks_url":"https://api.github.com/repos/Alluxio/alluxio/forks","keys_url":"https://api.github.com/repos/Alluxio/alluxio/keys{/key_id}","collaborators_url":"https://api.github.com/repos/Alluxio/alluxio/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/Alluxio/alluxio/teams","hooks_url":"https://api.github.com/repos/Alluxio/alluxio/hooks","issue_events_url":"https://api.github.com/repos/Alluxio/alluxio/issues/events{/number}","events_url":"https://api.github.com/repos/Alluxio/alluxio/events","assignees_url":"https://api.github.com/repos/Alluxio/alluxio/assignees{/user}","branches_url":"https://api.github.com/repos/Alluxio/alluxio/branches{/branch}","tags_url":"https://api.github.com/repos/Alluxio/alluxio/tags","blobs_url":"https://api.github.com/repos/Alluxio/alluxio/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/Alluxio/alluxio/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/Alluxio/alluxio/git/refs{/sha}","trees_url":"https://api.github.com/repos/Alluxio/alluxio/git/trees{/sha}","statuses_url":"https://api.github.com/repos/Alluxio/alluxio/statuses/{sha}","languages_url":"https://api.github.com/repos/Alluxio/alluxio/languages","stargazers_url":"https://api.github.com/repos/Alluxio/alluxio/stargazers","contributors_url":"https://api.github.com/repos/Alluxio/alluxio/contributors","subscribers_url":"https://api.github.com/repos/Alluxio/alluxio/subscribers","subscription_url":"https://api.github.com/repos/Alluxio/alluxio/subscription","commits_url":"https://api.github.com/repos/Alluxio/alluxio/commits{/sha}","git_commits_url":"https://api.github.com/repos/Alluxio/alluxio/git/commits{/sha}","comments_url":"https://api.github.com/repos/Alluxio/alluxio/comments{/number}","issue_comment_url":"https://api.github.com/repos/Alluxio/alluxio/issues/comments{/number}","contents_url":"https://api.github.com/repos/Alluxio/alluxio/contents/{+path}","compare_url":"https://api.github.com/repos/Alluxio/alluxio/compare/{base}...{head}","merges_url":"https://api.github.com/repos/Alluxio/alluxio/merges","archive_url":"https://api.github.com/repos/Alluxio/alluxio/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/Alluxio/alluxio/downloads","issues_url":"https://api.github.com/repos/Alluxio/alluxio/issues{/number}","pulls_url":"https://api.github.com/repos/Alluxio/alluxio/pulls{/number}","milestones_url":"https://api.github.com/repos/Alluxio/alluxio/milestones{/number}","notifications_url":"https://api.github.com/repos/Alluxio/alluxio/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/Alluxio/alluxio/labels{/name}","releases_url":"https://api.github.com/repos/Alluxio/alluxio/releases{/id}","deployments_url":"https://api.github.com/repos/Alluxio/alluxio/deployments","created_at":"2012-12-21T17:43:46Z","updated_at":"2016-05-11T14:10:44Z","pushed_at":"2016-05-12T19:47:58Z","git_url":"git://github.com/Alluxio/alluxio.git","ssh_url":"git@github.com:Alluxio/alluxio.git","clone_url":"https://github.com/Alluxio/alluxio.git","svn_url":"https://github.com/Alluxio/alluxio","homepage":"http://www.alluxio.org/","size":80788,"stargazers_count":2332,"watchers_count":2332,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":918,"mirror_url":null,"open_issues_count":29,"forks":918,"open_issues":29,"watchers":2332,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208"},"html":{"href":"https://github.com/Alluxio/alluxio/pull/3208"},"issue":{"href":"https://api.github.com/repos/Alluxio/alluxio/issues/3208"},"comments":{"href":"https://api.github.com/repos/Alluxio/alluxio/issues/3208/comments"},"review_comments":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208/comments"},"review_comment":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/Alluxio/alluxio/pulls/3208/commits"},"statuses":{"href":"https://api.github.com/repos/Alluxio/alluxio/statuses/ffa6381a9fed787b7e2b0933586cdd9e2f33597d"}}}}