Date: 2025-02-25

Status

Accepted

Context

As part of our migration from Ruby to Elixir, we need to leverage Elixir's concurrency model to improve the performance and reliability of our ebook format generation process. The current Ruby implementation processes formats sequentially, which leads to:

1. Longer build times: Each format (PDF, EPUB, MOBI, etc.) is generated one after another
2. Inefficient resource utilization: CPU cores remain idle during I/O-bound operations
3. Blocking operations: Long-running format generations block other formats from starting
4. Limited fault isolation: Failures in one format can affect the entire build process
5. Poor progress reporting: Difficult to track progress of multiple formats simultaneously

Our replatforming goals specifically mention leveraging Elixir's concurrency, fault tolerance, and distribution capabilities. Format generation is one of the most resource-intensive and time-consuming parts of our workflow, making it an ideal candidate for concurrent processing.

Decision

We will implement the following concurrency patterns for format generation in our Elixir replatforming:

1. Task-Based Parallelism for Independent Formats

For formats that can be generated independently (e.g., PDF, EPUB), we will use Task-based concurrency:

defmodule Codexis.FormatGeneration do
  def generate_formats(manuscript, formats) do
    # Start concurrent generation for each format
    tasks = 
      formats
      |> Enum.map(fn format ->
        Task.async(fn -> generate_format(manuscript, format) end)
      end)
    
    # Wait for all tasks to complete with timeout
    Task.await_many(tasks, 30_000)
  end
  
  defp generate_format(manuscript, :pdf) do
    # PDF generation logic
    {:ok, pdf_path}
  end
  
  defp generate_format(manuscript, :epub) do
    # EPUB generation logic
    {:ok, epub_path}
  end
  
  # Other format generators...
end

2. Supervised Worker Pools for Resource Management

For resource-intensive format generation, we'll use pooled workers under supervision:

defmodule Codexis.FormatGeneration.WorkerSupervisor do
  use Supervisor
  
  def start_link(args) do
    Supervisor.start_link(__MODULE__, args, name: __MODULE__)
  end
  
  @impl true
  def init(_args) do
    pool_size = Application.get_env(:elib, :format_worker_pool_size, 4)
    
    children = 
      for i <- 1..pool_size do
        worker_id = :"format_worker_#{i}"
        Supervisor.child_spec(
          {Codexis.FormatGeneration.Worker, [name: worker_id]},
          id: worker_id
        )
      end
    
    Supervisor.init(children, strategy: :one_for_one)
  end
end

defmodule Codexis.FormatGeneration.Worker do
  use GenServer
  
  # Worker implementation...
end

3. Flow-Based Processing for Format Pipelines

For formats with sequential dependencies (e.g., EPUB → MOBI → KFX), we'll use Flow:

defmodule Codexis.FormatGeneration.Pipeline do
  alias Codexis.FormatGeneration.Converters
  
  def generate_kindle_formats(manuscript) do
    [{:epub, manuscript}]
    |> Flow.from_enumerable()
    |> Flow.map(fn {:epub, manuscript} -> 
      {:epub, Converters.generate_epub(manuscript)}
    end)
    |> Flow.map(fn {:epub, {:ok, epub_path}} -> 
      {:mobi, Converters.epub_to_mobi(epub_path)}
    end)
    |> Flow.map(fn {:mobi, {:ok, mobi_path}} -> 
      {:kfx, Converters.mobi_to_kfx(mobi_path)}
    end)
    |> Flow.run()
  end
end

4. Dynamic Supervision for On-Demand Format Generation

For user-requested format generation, we'll use dynamic supervision:

defmodule Codexis.FormatGeneration.DynamicSupervisor do
  use DynamicSupervisor
  
  def start_link(args) do
    DynamicSupervisor.start_link(__MODULE__, args, name: __MODULE__)
  end
  
  @impl true
  def init(_args) do
    DynamicSupervisor.init(strategy: :one_for_one)
  end
  
  def start_format_generation(manuscript, format) do
    child_spec = {Codexis.FormatGeneration.FormatJob, {manuscript, format}}
    DynamicSupervisor.start_child(__MODULE__, child_spec)
  end
end

5. GenStage for Backpressure and Flow Control

For complex format generation with multiple stages and backpressure needs:

defmodule Codexis.FormatGeneration.Pipeline do
  defmodule Producer do
    use GenStage
    
    def start_link(manuscripts) do
      GenStage.start_link(__MODULE__, manuscripts)
    end
    
    @impl true
    def init(manuscripts) do
      {:producer, manuscripts}
    end
    
    @impl true
    def handle_demand(demand, manuscripts) when demand > 0 do
      {to_process, remaining} = Enum.split(manuscripts, demand)
      {:noreply, to_process, remaining}
    end
  end
  
  defmodule Processor do
    use GenStage
    
    def start_link(format) do
      GenStage.start_link(__MODULE__, format)
    end
    
    @impl true
    def init(format) do
      {:producer_consumer, format}
    end
    
    @impl true
    def handle_events(manuscripts, _from, format) do
      processed = Enum.map(manuscripts, &process_manuscript(&1, format))
      {:noreply, processed, format}
    end
    
    defp process_manuscript(manuscript, format) do
      # Process manuscript into specified format
      {manuscript, format, result_path}
    end
  end
  
  defmodule Consumer do
    use GenStage
    
    def start_link(_) do
      GenStage.start_link(__MODULE__, :ok)
    end
    
    @impl true
    def init(:ok) do
      {:consumer, :ok}
    end
    
    @impl true
    def handle_events(processed_items, _from, state) do
      # Store results, notify completion, etc.
      {:noreply, [], state}
    end
  end
end

6. Progress Tracking with PubSub

For real-time progress reporting across concurrent format generation:

defmodule Codexis.FormatGeneration.ProgressTracker do
  use GenServer
  
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end
  
  @impl true
  def init(_) do
    {:ok, %{formats: %{}}}
  end
  
  def update_progress(format, percentage, status) do
    GenServer.cast(__MODULE__, {:update, format, percentage, status})
  end
  
  @impl true
  def handle_cast({:update, format, percentage, status}, state) do
    new_state = 
      put_in(state, [:formats, format], %{percentage: percentage, status: status})
    
    # Broadcast progress update
    Phoenix.PubSub.broadcast(
      Codexis.PubSub, 
      "format_progress", 
      {:progress, format, percentage, status}
    )
    
    {:noreply, new_state}
  end
end

Implementation Strategy

Migration from Ruby to Elixir

The current Ruby implementation typically looks like:

class FormatGenerator
  def generate_all_formats(manuscript)
    formats = [:pdf, :epub, :mobi, :html]
    results = {}
    
    formats.each do |format|
      begin
        results[format] = generate_format(manuscript, format)
      rescue => e
        results[format] = { error: e.message }
      end
    end
    
    results
  end
  
  def generate_format(manuscript, format)
    # Format-specific generation logic
  end
end

This will be migrated to Elixir using the Task-based approach:

defmodule Codexis.FormatGeneration do
  def generate_all_formats(manuscript) do
    [:pdf, :epub, :mobi, :html]
    |> Task.async_stream(
         fn format -> {format, generate_format(manuscript, format)} end,
         timeout: 60_000,
         on_timeout: :kill_task
       )
    |> Enum.reduce(%{}, fn {:ok, {format, result}}, acc ->
         Map.put(acc, format, result)
       end)
  end
  
  defp generate_format(manuscript, format) do
    try do
      # Format-specific generation logic
      {:ok, result_path}
    rescue
      e -> {:error, Exception.message(e)}
    end
  end
end

Resource Management

For resource-intensive operations like PDF generation, we'll implement rate limiting:

defmodule Codexis.FormatGeneration.RateLimiter do
  use GenServer
  
  @max_concurrent 2  # Maximum concurrent PDF generations
  
  def start_link(_) do
    GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
  end
  
  @impl true
  def init(:ok) do
    {:ok, %{running: 0, queue: :queue.new()}}
  end
  
  def run_limited(fun) do
    GenServer.call(__MODULE__, {:run, fun})
  end
  
  @impl true
  def handle_call({:run, fun}, from, %{running: running, queue: queue} = state) do
    if running < @max_concurrent do
      Task.start(fn ->
        try do
          result = fun.()
          GenServer.cast(__MODULE__, :completed)
          GenServer.reply(from, result)
        rescue
          e ->
            GenServer.cast(__MODULE__, :completed)
            GenServer.reply(from, {:error, e})
        end
      end)
      
      {:noreply, %{state | running: running + 1}}
    else
      {:noreply, %{state | queue: :queue.in({fun, from}, queue)}}
    end
  end
  
  @impl true
  def handle_cast(:completed, %{running: running, queue: queue} = state) do
    case :queue.out(queue) do
      {{:value, {fun, from}}, new_queue} ->
        Task.start(fn ->
          try do
            result = fun.()
            GenServer.cast(__MODULE__, :completed)
            GenServer.reply(from, result)
          rescue
            e ->
              GenServer.cast(__MODULE__, :completed)
              GenServer.reply(from, {:error, e})
          end
        end)
        
        {:noreply, %{state | queue: new_queue}}
        
      {:empty, _} ->
        {:noreply, %{state | running: running - 1}}
    end
  end
end

Error Handling and Recovery

For robust error handling in concurrent format generation:

defmodule Codexis.FormatGeneration.Supervisor do
  use Supervisor
  
  def start_link(args) do
    Supervisor.start_link(__MODULE__, args, name: __MODULE__)
  end
  
  @impl true
  def init(_args) do
    children = [
      {Codexis.FormatGeneration.RateLimiter, []},
      {Codexis.FormatGeneration.ProgressTracker, []},
      {Codexis.FormatGeneration.WorkerSupervisor, []},
      {Codexis.FormatGeneration.ErrorHandler, []}
    ]
    
    Supervisor.init(children, strategy: :one_for_one)
  end
end

defmodule Codexis.FormatGeneration.ErrorHandler do
  use GenServer
  
  # Error handling implementation...
end

Consequences

Positive

1. Improved performance: Parallel format generation reduces overall build time
2. Better resource utilization: CPU cores are utilized efficiently
3. Enhanced fault tolerance: Failures in one format don't affect others
4. Scalability: System can scale to handle more formats or larger documents
5. Responsive feedback: Real-time progress reporting for each format
6. Graceful degradation: System can continue with available formats if some fail
7. Resource control: Prevents system overload during intensive operations
8. Simplified recovery: Easier to retry failed formats without reprocessing successful ones

Negative

1. Increased complexity: Concurrent code is more complex to understand and debug
2. Resource contention: Parallel processing may cause resource contention
3. Testing challenges: Concurrent code requires specialized testing approaches
4. Race conditions: Potential for subtle race conditions in format dependencies
5. Memory usage: Higher memory requirements for parallel processing
6. Coordination overhead: Need to manage and coordinate multiple concurrent processes

Mitigation Strategies

1. Resource pooling: Use worker pools to limit concurrent resource-intensive operations
2. Circuit breakers: Implement circuit breakers to prevent cascading failures
3. Backpressure mechanisms: Use GenStage for controlled flow of work
4. Supervision strategies: Implement proper supervision for fault tolerance
5. Timeout handling: Set appropriate timeouts for long-running operations
6. Graceful degradation: Design system to work with partial results
7. Comprehensive logging: Implement detailed logging for debugging
8. Property-based testing: Use property-based testing for concurrent code

References

• José Valim, "Elixir in Action"
• Bruce Tate & James Edward Gray II, "Designing Elixir Systems with OTP"
• "The Little Elixir & OTP Guidebook" by Benjamin Tan Wei Hao
• Elixir documentation on Task, GenServer, and Supervisor
• Flow and GenStage documentation
• "Concurrent Data Processing in Elixir" by Svilen Gospodinov