{"id":2426,"date":"2012-03-23T01:22:03","date_gmt":"2012-03-23T00:22:03","guid":{"rendered":"http:\/\/www.gamlor.info\/wordpress\/?p=2426"},"modified":"2021-03-11T09:19:12","modified_gmt":"2021-03-11T08:19:12","slug":"async-file-io-with-akka-and-java-7","status":"publish","type":"post","link":"https:\/\/www.gamlor.info\/wordpress\/2012\/03\/async-file-io-with-akka-and-java-7\/","title":{"rendered":"Async File IO with Akka and Java 7"},"content":{"rendered":"<p>Akka provides tons of nice facilities to deal with concurrent and asynchronous operations. However at the edges it often gets rougher when you deal with the non Akka world. For example traditional Java file access is synchronous. That can be annoying when you read large files, especially if those files are on another machine.<\/p>\n<p>Java 7 brings a new nice API for doing asynchronous file operations: The <a href=\"http:\/\/docs.oracle.com\/javase\/7\/docs\/api\/java\/nio\/channels\/AsynchronousFileChannel.html\">AsynchronousFileChannel<\/a> API. This allows us to read and write stuff asynchronously easily. I\u2019ve written a small wrapper which integrates a little better with Akka.<br \/>\n<div id=\"attachment_2439\" style=\"width: 288px\" class=\"wp-caption aligncenter\"><a href=\"https:\/\/www.gamlor.info\/wordpress\/wp-content\/uploads\/2012\/03\/akka-async-io.png\"><img loading=\"lazy\" decoding=\"async\" aria-describedby=\"caption-attachment-2439\" src=\"https:\/\/www.gamlor.info\/wordpress\/wp-content\/uploads\/2012\/03\/akka-async-io-278x300.png\" alt=\"Really, don&#039;t wait while the guy is reading\" title=\"akka-async-io\" width=\"278\" height=\"300\" class=\"size-medium wp-image-2439\" srcset=\"https:\/\/www.gamlor.info\/wordpress\/wp-content\/uploads\/2012\/03\/akka-async-io-278x300.png 278w, https:\/\/www.gamlor.info\/wordpress\/wp-content\/uploads\/2012\/03\/akka-async-io.png 600w\" sizes=\"(max-width: 278px) 100vw, 278px\" \/><\/a><p id=\"caption-attachment-2439\" class=\"wp-caption-text\">Really, don&#039;t wait while the guy is reading<\/p><\/div><\/p>\n<h2>Reading a File<\/h2>\n<p>First, the operations need an ExecutionContext, like <a href=\"http:\/\/doc.akka.io\/docs\/akka\/2.0\/scala\/futures.html\">other Akka constructs<\/a>. The execution context is used to dispatch the results of asynchronous request. So we need to declare an execution context. Actors already have an execution context which you can use for this. After that you can open files with the FileIO.open method. All operation will return there result as a Akka future, so that you can all tools from the Akka world:<\/p>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=BasicRead.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">import akka.dispatch.{ ExecutionContext, Promise }\nimport info.gamlor.io.FileIO\n \n\/\/ A plain execution context\nimplicit val dispatcher = ExecutionContext.fromExecutorService(yourExecutorServiceGoesHere)\n\n\/\/ or within an actor\nimplicit val dispatcher = context.dispatcher\n\nval file = FileIO.open(&quot;myFile.data&quot;)\n\/\/ read 200 bytes from the beginning of the file\nval readResultFuture = file.read(0,200)\n\n\/\/ do stuff with the future\nreadResultFuture.onSuccess({\n  case bytes:ByteString=&gt;{\n\tprintln(bytes.utf8String)\n  }\n}).andThen{ case _ =&gt; file.close()} \/\/ Close when we&#039;re done reading\n<\/code><\/pre><\/noscript>\n<p>The API wraps results in Akka <a href=\"http:\/\/doc.akka.io\/docs\/akka\/2.0\/scala\/io.html\">ByteStrings<\/a>, which are immutable byte arrays. This means you can easily send the raw data around and don\u2019t have to worry about any modifications.<\/p>\n<h2>Using Iteratees<\/h2>\n<p>Now when you read a file asynchronously data is usually transferred in chunks. That makes it harder to parse the data. One way to do this is with Iteratees. The FileIO instances can use Akka iteratees to process the file input. There two methods for this: The \u2018readAll\u2019 method reads the file until the Iteratee is done or the file ends. This is useful when you want to parse the file as a whole in a certain structure. The &#8216;readSegments&#8217; method reads until the iteratee is done, collects that result and start over parsing the rest of the file. This is handy when you need to parse a file with a repeated structure, for example parse each line.<br \/>\nParsing everything with an Iteratee:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=ParsingAll.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">case class Page(header: String, body: String, footer: String)\n\n\nval parser = for {\n\t  headerLine &lt;- IO.takeUntil(ByteString(&quot;\\n&quot;))\n\t  body &lt;- IO.takeUntil(ByteString(&quot;[End-Body]&quot;))\n\t  footer &lt;- IO.takeUntil(ByteString(&quot;[End-Footer]&quot;))\n} yield Page(headerLine.utf8String, body.utf8String, footer.utf8String)\n\/\/ We can use an iteraree as parser.\n\/\/ The parse result will be in the future.\n\/\/ There are overload available to read from certain positions.\nval readResultFuture = file.readAll(parser)\n\nreadResultFuture.onSuccess {\n  case Page(header, body, footer) =&gt; {\n\tprintln(header)\n\tprintln(body)\n\tprintln(footer)\n  }\n}<\/code><\/pre><\/noscript><br \/>\nParsing segments with an Iteratee:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=ParsingLines.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">case class LineItem(number: Int, content:String)\n   \nval parser = for {\n\t  numberOfItem &lt;- IO.takeUntil(ByteString(&quot;:&quot;))\n\t  lineContent &lt;- IO.takeUntil(ByteString(&quot;\\n&quot;))\n} yield LineItem(Integer.parseInt(numberOfItem.utf8String), lineContent.utf8String)\n\n\/\/ We can use an iteraree as parser.\n\/\/ This keep parsing until file ends\/max amount is reached.\n\/\/ Every time the iteraree is done parsing it will add that\n\/\/ to the result.\n\/\/ There are overload available to read from certain positions.\nval readResultFuture = file.readSegments(parser)\n\nreadResultFuture.onSuccess {\n  case items:Seq[LineItem] =&gt; {\n\titems.foreach({i=&gt;\n\t  println(i.number)\n\t  println(i.content)\n\t})\n  }\n}<\/code><\/pre><\/noscript><\/p>\n<h2>Reading Text<\/h2>\n<p>For text file you also can use the openText method. That one will return a TextIO-instance, which has text reading utilities like reading lines, split by some delimiter etc. The default encoding is UTF8 for these operations:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=ReadText.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">val textFile = FileIO.openText(&quot;lines.txt&quot;)\n\nval allLinesFuture = textFile.readAllLines()\n\nallLinesFuture.onSuccess({\n  case line:Seq[String]=&gt;{\n\tprintln(line)\n  }\n}).andThen{ case _ =&gt; file.close()}<\/code><\/pre><\/noscript><\/p>\n<h2>Writing to Files<\/h2>\n<p>In order to write to a file you need to open it with enough permissions. After that you can use the write methods. The write takes immutable ByteStrings and writes those to the file. Additional overloads also accept pure Java arrays and Java ByteBuffer. Those are mutable, so be very careful with those. If you mutate them during a asynchronous operation Dragons will appear:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=WriteToFile.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">val file = FileIO.open(Paths.get(&quot;myFile.data&quot;),StandardOpenOption.CREATE,StandardOpenOption.WRITE,StandardOpenOption.READ)\n\nfile.write(ByteString(&quot;data data&quot;),0).onComplete{\n  _=&gt;file.close()\n}<\/code><\/pre><\/noscript><\/p>\n<h2>Closing the Channel Automatically<\/h2>\n<p>So far we&#8217;ve always closed the file with an explicit call. Now when do you close the file? Because simply doing it in the finally clause doesn&#8217;t work, since the operation is running asynchronously in the background. A good option is to close the file when the last operation has finished.<br \/>\nFor this there is a &#8216;withFile&#8217; and a &#8216;withTextFile&#8217; method. This method accepts a closure which uses the given file and returns a future. When that future is completed the file will be closed. This means you can do complex file reading operations in that closure, return the result as a future and don;t worry about closing the file:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=AutoClose.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">case class LineItem(number: Int, content: String)\n\nval parser = for {\n  numberOfItem &lt;- IO.takeUntil(ByteString(&quot;:&quot;))\n  lineContent &lt;- IO.takeUntil(ByteString(&quot;\\n&quot;))\n} yield LineItem(Integer.parseInt(numberOfItem.utf8String), lineContent.utf8String)\n\n\/\/ will close the file when returned future has finished\nval onlyCoolLines = FileIO.withFile(Paths.get(&quot;aFile.txt&quot;)) {\n  file =&gt;\n\tval linesFuture = file.readSegments(parser)\n\tval coolLinesFuture = linesFuture.map(\n\t  lines =&gt; lines.filter(\n\t\tline =&gt; line.content.contains(&quot;cool&quot;)))\n\tcoolLinesFuture\n}\n\nonlyCoolLines.onSuccess {\n  case LineItem(no, line) =&gt; println(line)\n}<\/code><\/pre><\/noscript><\/p>\n<h2>IO Actors<\/h2>\n<p>For fault tolerance you try to move &#8216;dangerous&#8217; operations into separate actors, so that you can supervise those. The IOActor is intended for that. It does the asynchronous operations for you. If any operation goes wrong, it will fail with a IOException. It leaves the decision what to do to its supervisor. Here&#8217;s an example of an actor which reads pictures:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=ReadingPictures.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">class PictureIO extends Actor {\n\n\timport info.gamlor.io.IOActors._\n\n\n\toverride val supervisorStrategy = OneForOneStrategy(5, Duration(60, TimeUnit.SECONDS)) {\n\t  case ex: IOException =&gt; {\n\t\tprintln(&quot;Couldn't read file. Giving up on this file &quot;+ex)\n\t\tStop\n\t  }\n\t  case ex: Exception =&gt; Escalate\n\t}\n\n\tprotected def receive = {\n\t  case path: Path =&gt; {\n\t\t\/\/ will be created with the context,\n\t\t\/\/ so this actor supervises this file reading actor\n\t\tval fileReadingActor = IOActors.createForFile(path)\n\t\t\/\/ The actor will respond with a ReadResponse\n\t\t\/\/ which will contain the read data\n\t\tfileReadingActor ! Read(0, Int.MaxValue)\n\t  }\n\t  case ReadResponse(data, _, _) =&gt; {\n\t\tprocessTheBytesOfThisPicture(data)\n\t  }\n\t}\n\n\tdef processTheBytesOfThisPicture(data: ByteString) {\n\t  \/\/ do something\n\t}\n}\n<\/code><\/pre><\/noscript><\/p>\n<p>By default the IOActor will close the file after 5 seconds of inactivity. You can set that timeout when creating it. Or completely deactivate it and close the file by either stopping the actor or sending a &#8216;Close&#8217; message.<\/p>\n<h2>Chunked Reads with the IO Actor<\/h2>\n<p>When you read large files, you don&#8217;t want to read everything at once. You either can manually issue multiple read requests. Or you can use the ReadInChunks message. Then the actor will respond with multiple ReadInChunksResponse answers. It will send such a ReadChunk as soon as it has filled his internal buffers. When everything is done a last ReadChunk is sent with an EOF message. This is similar to <a href=\"http:\/\/doc.akka.io\/docs\/akka\/2.0\/scala\/io.html\">Akka&#8217;s network API<\/a>. So our picture reading actor would look something like this:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=ReadInChunks.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">protected def receive = {\n  case path: Path =&gt; {\n\t\/\/ will be created with the context,\n\t\/\/ so this actor supervises this file reading actor\n\tval fileReadingActor = IOActors.createForFile(path)\n\t\/\/ The actor will respond with a ReadResponse\n\t\/\/ which will contain the read data\n\tfileReadingActor ! ReadInChunks(0, Int.MaxValue,path)\n  }\n  case ReadInChunksResponse(data, _) =&gt; {\n\tdata match{\n\t  case IO.Chunk(bytes) =&gt;processPartOfPicture(bytes)\n\t  case IO.EOF =&gt; finishPicture()  \n\t}\n\t\n  }\n}\n\ndef processPartOfPicture(string: ByteString){\n  \/\/ do stuff\n}\n\ndef finishPicture(){\n  \/\/ done\n}<\/code><\/pre><\/noscript><\/p>\n<h2>Buffer-Management<\/h2>\n<p>Well, async IO operations need to allocate a Java ByteBuffer to transfer the data. Currently this buffer management is as simple as possible. It allocates a buffer for every operation you start. So every time you call a read or write method it will allocate a buffer for that. Of course when you write a very large file with one call it will use the same buffer during that operation.<br \/>\nThis is good enough for now =). In general: I haven&#8217;t done any decent performance testing yet.<\/p>\n<h2>Getting This Stuff<\/h2>\n<p>First: This stuff is not yet written in stone and the API may change. Anyway: The code is on <a href=\"https:\/\/github.com\/gamlerhart\/akka-async-apis\">GitHub<\/a>. Also I&#8217;ve pushed SNAPSHOTS to a GitHub based Maven repository.<\/p>\n<p>Repository for Maven: https:\/\/github.com\/gamlerhart\/gamlor-mvn\/raw\/master\/snapshots<br \/>\nGroupID: info.gamlor.akkaasync<br \/>\nArtifactID: akka-io_2.9.1<br \/>\nVersion: 1.0-SNAPSHOT<\/p>\n<p>For example in SBT:<br \/>\n<script src=\"https:\/\/gist.github.com\/2165529.js?file=GettingItWithSBT.scala\"><\/script><noscript><pre><code class=\"language-scala scala\">resolvers += &quot;Gamlor-Repo&quot; at &quot;https:\/\/github.com\/gamlerhart\/gamlor-mvn\/raw\/master\/snapshots&quot;\n\nlibraryDependencies += &quot;com.typesafe.akka&quot; % &quot;akka-actor&quot; % &quot;2.0&quot;\nlibraryDependencies += &quot;info.gamlor.akkaasync&quot;  %% &quot;akka-io&quot; % &quot;1.0-SNAPSHOT&quot;<\/code><\/pre><\/noscript><\/p>\n<h2>That&#8217;s It<\/h2>\n<p>Improvements will certainly follow. Also I&#8217;ve a thin wrapper for a the Async Http Client in the works That&#8217;s the topic for a follow up post =)<\/p>\n","protected":false},"excerpt":{"rendered":"<p>Akka provides tons of nice facilities to deal with concurrent and asynchronous operations. However at the edges it often gets rougher when you deal with the non Akka world. For&hellip; <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":"","_links_to":"","_links_to_target":""},"categories":[15,17],"tags":[245,263,226],"aioseo_notices":[],"_links":{"self":[{"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/posts\/2426"}],"collection":[{"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/comments?post=2426"}],"version-history":[{"count":20,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/posts\/2426\/revisions"}],"predecessor-version":[{"id":3756,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/posts\/2426\/revisions\/3756"}],"wp:attachment":[{"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/media?parent=2426"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/categories?post=2426"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.gamlor.info\/wordpress\/wp-json\/wp\/v2\/tags?post=2426"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}