Support distributed File.Stream (#12578)

Author: josevalim

Makefile

@@ -281,9 +281,9 @@ test_stdlib: compile
 	@ echo "==> elixir (ex_unit)"
 	$(Q) exec epmd & exit
 	$(Q) if [ "$(OS)" = "Windows_NT" ]; then \
-		cd lib/elixir && cmd //C call ../../bin/elixir.bat -r "test/elixir/test_helper.exs" -pr "test/elixir/**/$(TEST_FILES)"; \
+		cd lib/elixir && cmd //C call ../../bin/elixir.bat --sname primary -r "test/elixir/test_helper.exs" -pr "test/elixir/**/$(TEST_FILES)"; \
 	else \
-		cd lib/elixir && ../../bin/elixir -r "test/elixir/test_helper.exs" -pr "test/elixir/**/$(TEST_FILES)"; \
+		cd lib/elixir && ../../bin/elixir --sname primary -r "test/elixir/test_helper.exs" -pr "test/elixir/**/$(TEST_FILES)"; \
 	fi
 
 #==> Dialyzer tasks

lib/elixir/lib/file.ex

@@ -1682,6 +1682,11 @@ defmodule File do
   using the `:line` option, CRLF line breaks (`"\r\n"`) are normalized
   to LF (`"\n"`).
 
+  Similar to other file operations, a stream can be created in one node
+  and forwarded to another node. Once the stream is opened in another node,
+  a request will be sent to the creator node to spawn a process for file
+  streaming.
+
   Operating the stream can fail on open for the same reasons as
   `File.open!/2`. Note that the file is automatically opened each time streaming
   begins. There is no need to pass `:read` and `:write` modes, as those are
@@ -1692,12 +1697,13 @@ defmodule File do
   Since Elixir controls when the streamed file is opened, the underlying
   device cannot be shared and as such it is convenient to open the file
   in raw mode for performance reasons. Therefore, Elixir **will** open
-  streams in `:raw` mode with the `:read_ahead` option unless an encoding
-  is specified. This means any data streamed into the file must be
-  converted to `t:iodata/0` type. If you pass, for example, `[encoding: :utf8]`
-  or `[encoding: {:utf16, :little}]` in the modes parameter,
-  the underlying stream will use `IO.write/2` and the `String.Chars` protocol
-  to convert the data. See `IO.binwrite/2` and `IO.write/2` .
+  streams in `:raw` mode with the `:read_ahead` option if the stream is
+  open in the same node as it is created and no encoding has been specified.
+  This means any data streamed into the file must be converted to `t:iodata/0`
+  type. If you pass, for example, `[encoding: :utf8]` or
+  `[encoding: {:utf16, :little}]` in the modes parameter, the underlying stream
+  will use `IO.write/2` and the `String.Chars` protocol to convert the data.
+  See `IO.binwrite/2` and `IO.write/2` .
 
   One may also consider passing the `:delayed_write` option if the stream
   is meant to be written to under a tight loop.
@@ -1707,8 +1713,8 @@ defmodule File do
   If you pass `:trim_bom` in the modes parameter, the stream will
   trim UTF-8, UTF-16 and UTF-32 byte order marks when reading from file.
 
-  Note that this function does not try to discover the file encoding basing
-  on BOM.
+  Note that this function does not try to discover the file encoding
+  based on BOM.
 
   ## Examples
 

lib/elixir/lib/file/stream.ex

@@ -8,10 +8,11 @@ defmodule File.Stream do
     * `modes`         - the file modes
     * `raw`           - a boolean indicating if bin functions should be used
     * `line_or_bytes` - if reading should read lines or a given number of bytes
+    * `node`          - the node the file belongs to
 
   """
 
-  defstruct path: nil, modes: [], line_or_bytes: :line, raw: true
+  defstruct path: nil, modes: [], line_or_bytes: :line, raw: true, node: nil
 
   @type t :: %__MODULE__{}
 
@@ -32,19 +33,29 @@ defmodule File.Stream do
           modes
       end
 
-    %File.Stream{path: path, modes: modes, raw: raw, line_or_bytes: line_or_bytes}
+    %File.Stream{path: path, modes: modes, raw: raw, line_or_bytes: line_or_bytes, node: node()}
+  end
+
+  @doc false
+  def __open__(%File.Stream{path: path, node: node}, modes) when node == node() do
+    :file.open(path, modes)
+  end
+
+  @doc false
+  def __open__(%File.Stream{path: path, node: node}, modes) do
+    :erpc.call(node, :file_io_server, :start, [self(), path, List.delete(modes, :raw)])
   end
 
   defimpl Collectable do
-    def into(%{path: path, modes: modes, raw: raw} = stream) do
+    def into(%{modes: modes, raw: raw} = stream) do
       modes = for mode <- modes, mode not in [:read], do: mode
 
-      case :file.open(path, [:write | modes]) do
+      case File.Stream.__open__(stream, [:write | modes]) do
         {:ok, device} ->
           {:ok, into(device, stream, raw)}
 
         {:error, reason} ->
-          raise File.Error, reason: reason, action: "stream", path: path
+          raise File.Error, reason: reason, action: "stream", path: stream.path
       end
     end
 
@@ -73,14 +84,14 @@ defmodule File.Stream do
   defimpl Enumerable do
     @read_ahead_size 64 * 1024
 
-    def reduce(%{path: path, modes: modes, line_or_bytes: line_or_bytes, raw: raw}, acc, fun) do
+    def reduce(%{modes: modes, line_or_bytes: line_or_bytes, raw: raw} = stream, acc, fun) do
       start_fun = fn ->
-        case :file.open(path, read_modes(modes)) do
+        case File.Stream.__open__(stream, read_modes(modes)) do
           {:ok, device} ->
             if :trim_bom in modes, do: trim_bom(device, raw) |> elem(0), else: device
 
           {:error, reason} ->
-            raise File.Error, reason: reason, action: "stream", path: path
+            raise File.Error, reason: reason, action: "stream", path: stream.path
         end
       end
 
@@ -93,27 +104,20 @@ defmodule File.Stream do
       Stream.resource(start_fun, next_fun, &:file.close/1).(acc, fun)
     end
 
-    def count(%{path: path, modes: modes, line_or_bytes: :line} = stream) do
+    def count(%{modes: modes, line_or_bytes: :line, path: path} = stream) do
       pattern = :binary.compile_pattern("\n")
       counter = &count_lines(&1, path, pattern, read_function(stream), 0)
-
-      case File.open(path, read_modes(modes), counter) do
-        {:ok, count} ->
-          {:ok, count}
-
-        {:error, reason} ->
-          raise File.Error, reason: reason, action: "stream", path: path
-      end
+      {:ok, open!(stream, modes, counter)}
     end
 
-    def count(%{path: path, line_or_bytes: bytes, raw: true, modes: modes}) do
-      case File.stat(path) do
+    def count(%{path: path, line_or_bytes: bytes, raw: true, modes: modes, node: node} = stream) do
+      case :erpc.call(node, File, :stat, [path]) do
         {:ok, %{size: 0}} ->
           {:error, __MODULE__}
 
         {:ok, %{size: size}} ->
           remainder = if rem(size, bytes) == 0, do: 0, else: 1
-          {:ok, div(size, bytes) + remainder - count_raw_bom(path, modes)}
+          {:ok, div(size, bytes) + remainder - count_raw_bom(stream, modes)}
 
         {:error, reason} ->
           raise File.Error, reason: reason, action: "stream", path: path
@@ -132,9 +136,23 @@ defmodule File.Stream do
       {:error, __MODULE__}
     end
 
-    defp count_raw_bom(path, modes) do
+    defp open!(stream, modes, fun) do
+      case File.Stream.__open__(stream, read_modes(modes)) do
+        {:ok, device} ->
+          try do
+            fun.(device)
+          after
+            :file.close(device)
+          end
+
+        {:error, reason} ->
+          raise File.Error, reason: reason, action: "stream", path: stream.path
+      end
+    end
+
+    defp count_raw_bom(stream, modes) do
       if :trim_bom in modes do
-        File.open!(path, read_modes(modes), &(&1 |> trim_bom(true) |> elem(1)))
+        open!(stream, read_modes(modes), &(&1 |> trim_bom(true) |> elem(1)))
       else
         0
       end