{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/6586","labels_url":"https://api.github.com/repos/apache/spark/issues/6586/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/6586/comments","events_url":"https://api.github.com/repos/apache/spark/issues/6586/events","html_url":"https://github.com/apache/spark/pull/6586","id":84035504,"number":6586,"title":"[SPARK-8044][CORE]Avoid to make directbuffer out of control  while reading or droping disk level block","user":{"login":"suyanNone","id":9818852,"avatar_url":"https://avatars.githubusercontent.com/u/9818852?v=3","gravatar_id":"","url":"https://api.github.com/users/suyanNone","html_url":"https://github.com/suyanNone","followers_url":"https://api.github.com/users/suyanNone/followers","following_url":"https://api.github.com/users/suyanNone/following{/other_user}","gists_url":"https://api.github.com/users/suyanNone/gists{/gist_id}","starred_url":"https://api.github.com/users/suyanNone/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/suyanNone/subscriptions","organizations_url":"https://api.github.com/users/suyanNone/orgs","repos_url":"https://api.github.com/users/suyanNone/repos","events_url":"https://api.github.com/users/suyanNone/events{/privacy}","received_events_url":"https://api.github.com/users/suyanNone/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":19,"created_at":"2015-06-02T13:55:27Z","updated_at":"2015-06-04T09:22:20Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/6586","html_url":"https://github.com/apache/spark/pull/6586","diff_url":"https://github.com/apache/spark/pull/6586.diff","patch_url":"https://github.com/apache/spark/pull/6586.patch"},"body":"1.  I found if we use getChannel to put or get data, it will create DirectBuffer anyway, which is not controllable.\r\n\r\naccording openJDK source code: because it will create a ThreadLocal directBuffer pool, and is not provider a 100% percent way to sure the direct buffer to be released.\r\n\r\n```\r\n sun.nio.ch.FileChannelImpl.java\r\npublic int write(ByteBuffer src) throws IOException {\r\n210         ensureOpen();\r\n211         if (!writable)\r\n212             throw new NonWritableChannelException();\r\n213         synchronized (positionLock) {\r\n214             int n = 0;\r\n215             int ti = -1;\r\n216             try {\r\n217                 begin();\r\n218                 if (!isOpen())\r\n219                     return 0;\r\n220                 ti = threads.add();\r\n221                 if (appending)\r\n222                     position(size());\r\n223                 do {\r\n224                     n = IOUtil.write(fd, src, -1, nd, positionLock);\r\n225                 } while ((n == IOStatus.INTERRUPTED) && isOpen());\r\n226                 return IOStatus.normalize(n);\r\n227             } finally {\r\n228                 threads.remove(ti);\r\n229                 end(n > 0);\r\n230                 assert IOStatus.check(n);\r\n231             }\r\n232         }\r\n233     }\r\n```\r\n```\r\nIOUtil.java\r\n\r\nstatic int write(FileDescriptor fd, ByteBuffer src, long position,\r\n74                      NativeDispatcher nd, Object lock)\r\n75         throws IOException\r\n76     {\r\n77         if (src instanceof DirectBuffer)\r\n78             return writeFromNativeBuffer(fd, src, position, nd, lock);\r\n79 \r\n80         // Substitute a native buffer\r\n81         int pos = src.position();\r\n82         int lim = src.limit();\r\n83         assert (pos <= lim);\r\n84         int rem = (pos <= lim ? lim - pos : 0);\r\n85         ByteBuffer bb = null;\r\n86         try {\r\n87             bb = Util.getTemporaryDirectBuffer(rem);\r\n88             bb.put(src);\r\n89             bb.flip();\r\n90             // Do not update src until we see how many bytes were written\r\n91             src.position(pos);\r\n92 \r\n93             int n = writeFromNativeBuffer(fd, bb, position, nd, lock);\r\n94             if (n > 0) {\r\n95                 // now update src\r\n96                 src.position(pos + n);\r\n97             }\r\n98             return n;\r\n99         } finally {\r\n100            Util.releaseTemporaryDirectBuffer(bb);\r\n101        }\r\n102    }\r\n```\r\n```\r\nUtil.java\r\n     static ByteBuffer getTemporaryDirectBuffer(int size) {\r\n61         ByteBuffer buf = null;\r\n62         // Grab a buffer if available\r\n63         for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {\r\n64             SoftReference ref = (SoftReference)(bufferPool[i].get());\r\n65             if ((ref != null) && ((buf = (ByteBuffer)ref.get()) != null) &&\r\n66                 (buf.capacity() >= size)) {\r\n67                 buf.rewind();\r\n68                 buf.limit(size);\r\n69                 bufferPool[i].set(null);\r\n70                 return buf;\r\n71             }\r\n72         }\r\n73 \r\n74         // Make a new one\r\n75         return ByteBuffer.allocateDirect(size);\r\n76     }\r\n```\r\n```\r\n private static final int TEMP_BUF_POOL_SIZE = 3;\r\n50 \r\n51     // Per-thread soft cache of the last temporary direct buffer\r\n52     private static ThreadLocal[] bufferPool;\r\n53 \r\n54     static {\r\n55         bufferPool = new ThreadLocal[TEMP_BUF_POOL_SIZE];\r\n56         for (int i=0; i<TEMP_BUF_POOL_SIZE; i++)\r\n57             bufferPool[i] = new ThreadLocal();\r\n58     }\r\n59 \r\n60     static ByteBuffer getTemporaryDirectBuffer(int size) {\r\n61         ByteBuffer buf = null;\r\n62         // Grab a buffer if available\r\n63         for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {\r\n64             SoftReference ref = (SoftReference)(bufferPool[i].get());\r\n65             if ((ref != null) && ((buf = (ByteBuffer)ref.get()) != null) &&\r\n66                 (buf.capacity() >= size)) {\r\n67                 buf.rewind();\r\n68                 buf.limit(size);\r\n69                 bufferPool[i].set(null);\r\n70                 return buf;\r\n71             }\r\n72         }\r\n73 \r\n74         // Make a new one\r\n75         return ByteBuffer.allocateDirect(size);\r\n76     }\r\n77 \r\n78     static void releaseTemporaryDirectBuffer(ByteBuffer buf) {\r\n79         if (buf == null)\r\n80             return;\r\n81         // Put it in an empty slot if such exists\r\n82         for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {\r\n83             SoftReference ref = (SoftReference)(bufferPool[i].get());\r\n84             if ((ref == null) || (ref.get() == null)) {\r\n85                 bufferPool[i].set(new SoftReference(buf));\r\n86                 return;\r\n87             }\r\n88         }\r\n89         // Otherwise replace a smaller one in the cache if such exists\r\n90         for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {\r\n91             SoftReference ref = (SoftReference)(bufferPool[i].get());\r\n92             ByteBuffer inCacheBuf = (ByteBuffer)ref.get();\r\n93             if ((inCacheBuf == null) || (buf.capacity() > inCacheBuf.capacity())) {\r\n94                 bufferPool[i].set(new SoftReference(buf));\r\n95                 return;\r\n96             }\r\n97         }\r\n98     }\r\n\r\n```"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/108802020","html_url":"https://github.com/apache/spark/pull/6586#issuecomment-108802020","issue_url":"https://api.github.com/repos/apache/spark/issues/6586","id":108802020,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2015-06-04T09:22:20Z","updated_at":"2015-06-04T09:22:20Z","body":"Merged build started."}}