ruby-celluloid 0.16.0-4 / usr / lib / ruby / vendor_ruby / celluloid / tasks.rb

module Celluloid
  # Asked to do task-related things outside a task
  class NotTaskError < Celluloid::Error; end

  # Trying to resume a dead task
  class DeadTaskError < Celluloid::Error; end

  # Errors which should be resumed automatically
  class ResumableError < Celluloid::Error; end

  # Tasks are interruptable/resumable execution contexts used to run methods
  class Task
    class TerminatedError < ResumableError; end # kill a running task after terminate

    class TimeoutError < ResumableError; end # kill a running task after timeout

    # Obtain the current task
    def self.current
      Thread.current[:celluloid_task] or raise NotTaskError, "not within a task context"
    end

    # Suspend the running task, deferring to the scheduler
    def self.suspend(status)
      Task.current.suspend(status)
    end

    attr_reader :type, :meta, :status
    attr_accessor :chain_id, :guard_warnings

    # Create a new task
    def initialize(type, meta)
      @type     = type
      @meta     = meta
      @status   = :new

      @exclusive         = false
      @dangerous_suspend = @meta ? @meta.delete(:dangerous_suspend) : false
      @guard_warnings    = false

      actor     = Thread.current[:celluloid_actor]
      @chain_id = CallChain.current_id

      raise NotActorError, "can't create tasks outside of actors" unless actor
      guard "can't create tasks inside of tasks" if Thread.current[:celluloid_task]

      create do
        begin
          @status = :running
          actor.setup_thread

          name_current_thread thread_metadata

          Thread.current[:celluloid_task] = self
          CallChain.current_id = @chain_id

          actor.tasks << self
          yield
        rescue Task::TerminatedError
          # Task was explicitly terminated
        ensure
          name_current_thread nil
          @status = :dead
          actor.tasks.delete self
        end
      end
    end

    def create(&block)
      raise "Implement #{self.class}#create"
    end

    # Suspend the current task, changing the status to the given argument
    def suspend(status)
      raise "Cannot suspend while in exclusive mode" if exclusive?
      raise "Cannot suspend a task from outside of itself" unless Task.current == self

      @status = status

      if $CELLULOID_DEBUG && @dangerous_suspend
        Logger.with_backtrace(caller[2...8]) do |logger|
          logger.warn "Dangerously suspending task: type=#{@type.inspect}, meta=#{@meta.inspect}, status=#{@status.inspect}"
        end
      end

      value = signal

      @status = :running
      raise value if value.is_a?(Celluloid::ResumableError)

      value
    end

    # Resume a suspended task, giving it a value to return if needed
    def resume(value = nil)
      guard "Cannot resume a task from inside of a task" if Thread.current[:celluloid_task]
      deliver(value)
      nil
    end

    # Execute a code block in exclusive mode.
    def exclusive
      if @exclusive
        yield
      else
        begin
          @exclusive = true
          yield
        ensure
          @exclusive = false
        end
      end
    end

    # Terminate this task
    def terminate
      raise "Cannot terminate an exclusive task" if exclusive?

      if running?
        Logger.with_backtrace(backtrace) do |logger|
          logger.warn "Terminating task: type=#{@type.inspect}, meta=#{@meta.inspect}, status=#{@status.inspect}"
        end
        exception = Task::TerminatedError.new("task was terminated")
        exception.set_backtrace(caller)
        resume exception
      else
        raise DeadTaskError, "task is already dead"
      end
    end

    # Is this task running in exclusive mode?
    def exclusive?
      @exclusive
    end

    def backtrace
    end

    # Is the current task still running?
    def running?; @status != :dead; end

    # Nicer string inspect for tasks
    def inspect
      "#<#{self.class}:0x#{object_id.to_s(16)} @type=#{@type.inspect}, @meta=#{@meta.inspect}, @status=#{@status.inspect}>"
    end

    def guard(message)
      if @guard_warnings
        Logger.warn message if $CELLULOID_DEBUG
      else
        raise message if $CELLULOID_DEBUG
      end
    end

    private

    def name_current_thread(new_name)
      return unless RUBY_PLATFORM == "java"
      if new_name.nil?
        new_name = Thread.current[:celluloid_original_thread_name]
        Thread.current[:celluloid_original_thread_name] = nil
      else
        Thread.current[:celluloid_original_thread_name] = Thread.current.to_java.getNativeThread.get_name
      end
      Thread.current.to_java.getNativeThread.set_name(new_name)
    end

    def thread_metadata
      method = @meta && @meta[:method_name] || "<no method>"
      klass = Thread.current[:celluloid_actor] && Thread.current[:celluloid_actor].behavior.subject.bare_object.class || "<no actor>"
      format("[Celluloid] %s#%s", klass, method)
    end
  end
end

require 'celluloid/tasks/task_fiber'
require 'celluloid/tasks/task_thread'