123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::ProcessingActor

Overview

Note:

**Edge Features** are under active development and may change frequently.

  • Deprecations are not added before incompatible changes.

  • Edge version: major is always 0, minor bump means incompatible change, patch bump means compatible change.

  • Edge features may also lack tests and documentation.

  • Features developed in concurrent-ruby-edge are expected to move to concurrent-ruby when finalised.

A new implementation of actor which also simulates the process, therefore it can be used in the same way as Erlang’s actors but without occupying thread. A tens of thousands ProcessingActors can run at the same time sharing a thread pool.

Examples:

# Runs on a pool, does not consume 50_000 threads
actors = 50_000.times.map do |i|
  Concurrent::ProcessingActor.act(i) { |a, i| a.receive.then_on(:fast, i) { |m, i| m + i } }
end

actors.each { |a| a.tell 1 }
values = actors.map(&:termination).map(&:value)
values[0,5]                                        # => [1, 2, 3, 4, 5]
values[-5, 5]                                      # => [49996, 49997, 49998, 49999, 50000]

Class Attribute Summary

Synchronization::Object - Inherited

.safe_initialization?

Class Method Summary

Synchronization::Object - Inherited

.atomic_attribute?, .atomic_attributes,
.attr_atomic

Creates methods for reading and writing to a instance variable with volatile (Java) semantic as .attr_volatile does.
.attr_volatile

Creates methods for reading and writing (as attr_accessor does) to a instance variable with volatile (Java) semantic.
.ensure_safe_initialization_when_final_fields_are_present

For testing purposes, quite slow.
.new

Has to be called by children.
.safe_initialization!, .define_initialize_atomic_fields

Synchronization::AbstractObject - Inherited

.attr_volatile, .new

Instance Method Summary

Synchronization::Object - Inherited

#__initialize_atomic_fields__

Synchronization::Volatile - Included

#full_memory_barrier

Synchronization::AbstractObject - Inherited

#full_memory_barrier

Constructor Details

.new(channel, *args, &process) ⇒ ProcessingActor (private)

[ GitHub ] 

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 177



def initialize(channel, *args, &process)
  @Mailbox    = channel
  @Terminated = Promises.future(self, *args, &process).run
  super()
end


  







Class Method Details



  

    .act(*args, &process)  ⇒ ProcessingActor 
  



  

    

Creates an actor.


  





    

    
Examples:

      
actor = Concurrent::ProcessingActor.act do |actor|
  actor.receive.then do |message|
    # the actor ends normally with message
    message
  end
end

actor.tell :a_message
    # => <#Concurrent::ProcessingActor:0x7fff11280560 termination:pending>
actor.termination.value! # => :a_message

  

  
See Also:

  
    
      
  • Behaves the same way, but does not take mailbox as a first argument.
    • 
      




  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 54



def self.act(*args, &process)
  act_listening Promises::Channel.new, *args, &process
end


  








  

    .act_listening(channel, *args) {|actor, *args| ... } ⇒ ProcessingActor 
  



  

    

Creates an actor listening to a specified channel (mailbox).


  





  
Parameters:


    
  
  • 
    args
    (Object)
—    
    

    Arguments passed to the process.
    

    
  
    • 
  
  • 
    channel
    (Promises::Channel)
—    
    

    which serves as mailing box. The channel can have limited size to achieve backpressure.
    

    
  
    • 



Yields:


    
  
  • 
    (actor, *args)
—    
    

    to the process to get back a future which represents the actors execution.
    

    
  
    • 



Yield Parameters:


    
  
  • 
    actor
    (ProcessingActor)
  
    • 
  
  • 
    *args
    (Object)
  
    • 



Yield Returns:





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 67



def self.act_listening(channel, *args, &process)
  ProcessingActor.new channel, *args, &process
end


  







Instance Method Details



  

    #ask_op(answer = Promises.resolvable_future, &message_provider)  
  



  

    

actor.ask2 { |a| [:count, a] }


  





  


  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 157



def ask_op(answer = Promises.resolvable_future, &message_provider)
  # TODO (pitr-ch 12-Dec-2018): is it ok to let the answers be unanswered when the actor terminates
  tell_op(message_provider.call(answer)).then(answer) { |_, a| a }

  # answer.chain { |v| [true, v] } | @Terminated.then
end


  








  

    #inspect  
  



  

    

Alias for #to_s.


  



  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 169



alias_method :inspect, :to_s


  








  

    #mailbox  ⇒ Promises::Channel 
  



  

    
  





  
Returns:





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 29



def mailbox
  @Mailbox
end


  








  

    #receive(channel = mailbox)  
  



  

    

Receives a message when available, used in the actor’s process.



@return [Promises::Future(Object)] a future which will be fulfilled with a message from



mailbox when it is available.



def receive(*channels)



channels = [@Mailbox] if channels.empty?
Promises::Channel.select(*channels)
# TODO (pitr-ch 27-Dec-2016): support patterns
#   - put any received message aside if it does not match
#   - on each receive call check the messages put aside
#   - track where the message came from, cannot later receive m form other channel only because it matches



end


  





  


  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 83



def receive(channel = mailbox)
  channel.pop_op
end


  








  

    #tell!(message)  ⇒ self 
  



  

    

Tells a message to the actor. May block current thread if the mailbox is full. #tell_op is a better option since it does not block. It’s usually used to integrate with threading code.


  





    

    
Examples:

      
Thread.new(actor) do |actor|
  # ...
  actor.tell! :a_message # blocks until the message is told
  #   (there is a space for it in the channel)
  # ...
end

  


Parameters:





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 99



def tell!(message)
  @Mailbox.push(message)
  self
end


  








  

    #tell_op(message)  ⇒ Promises::Future(ProcessingActor) 
  



  

    

Tells a message to the actor.


  





  
Parameters:



Returns:


    
  
  • 
    (Promises::Future(ProcessingActor))
—    
    

    a future which will be fulfilled with the actor when the message is pushed to mailbox.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 108



def tell_op(message)
  @Mailbox.push_op(message).then(self) { |_ch, actor| actor }
end


  








  

    #termination  ⇒ Promises::Future(Object) 
  



  

    
  





  
Returns:


    
  
  • 
    (Promises::Future(Object))
—    
    

    a future which is resolved when the actor ends its processing. It can either be fulfilled with a value when actor ends normally or rejected with a reason (exception) when actor fails.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 36



def termination
  @Terminated.with_hidden_resolvable
end


  








  

    #to_ary  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 171



def to_ary
  [@Mailbox, @Terminated]
end


  








  

    #to_s  ⇒ String 
    Also known as: #inspect
  



  

    
  





  
Returns:


    
  
  • 
    (String)
—    
    

    string representation.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/edge/processing_actor.rb', line 165



def to_s
  format '%s termination: %s>', super[0..-2], termination.state
end