class Syskit::RemoteStateGetter

@api private

Adapter that reads the synchronous state of a remote node asynchronously

It is started in a paused state, waiting for {#resume} to be called to start polling the remote state

Attributes

current_state[R]

The newest known state

exit_condition[R]

The condition used to make the polling thread quit

last_read_state[R]

The last read state

orocos_task[R]

The task we're polling

period[R]

The polling period in seconds

poll_thread[R]

The thread that polls rtt_state

poll_thread_error[R]

The exception that terminated {#poll_thread}

poll_thread_exit_sync[R]

To avoid deadlocking in {#wait} if the thread quits

run_event[R]

Control for pause/resume

state_queue[R]

The queue of values read by poll_thread

sync_latch[R]

Holder for the latch that is used by {#wait} for synchronization

Public Class Methods

new(orocos_task, initial_state: nil, period: 0.02) click to toggle source

# File lib/syskit/remote_state_getter.rb, line 32
def initialize(orocos_task, initial_state: nil, period: 0.02)
    @orocos_task = orocos_task
    @period = period
    @exit_condition = Concurrent::Event.new
    @run_event = Concurrent::Event.new
    @current_state = Concurrent::Atom.new(initial_state)
    @state_queue = Queue.new
    @sync_latch = Concurrent::Atom.new(nil)
    @poll_thread_error = Concurrent::Atom.new(nil)
    @poll_thread_exit_sync = Mutex.new
    @last_read_state = nil
    if initial_state
        state_queue.push(initial_state)
    end

    period = self.period
    exit_condition.reset
    @poll_thread = Thread.new do
        poll_loop
    end
end

Public Instance Methods

clear() click to toggle source

# File lib/syskit/remote_state_getter.rb, line 143
def clear
    state_queue.clear
    current_state.reset(nil)
    @last_read_state = nil
end

connected?() click to toggle source

Whether the polling thread is alive

# File lib/syskit/remote_state_getter.rb, line 150
def connected?
    !exit_condition.set?
end

disconnect() click to toggle source

Stop polling completely

After calling {#disconnect}, the reader cannot be connected again. Use {#pause} and {#resume} to temporarily stop polling

# File lib/syskit/remote_state_getter.rb, line 170
def disconnect
    # This order ensures that {#poll_thread} will quit immediately if it
    # was paused
    exit_condition.set
    run_event.set
end

join() click to toggle source

Wait for the poll thread to finish after a {#disconnect}

# File lib/syskit/remote_state_getter.rb, line 178
def join
    poll_thread.join
end

pause() click to toggle source

Pause polling until {#resume} or {#disconnect} are called

# File lib/syskit/remote_state_getter.rb, line 155
def pause
    run_event.reset
end

poll_loop() click to toggle source

The internal polling loop

# File lib/syskit/remote_state_getter.rb, line 55
def poll_loop
    # Analysis to avoid deadlocking in {#wait}
    #
    # In the ensure block:
    # - we've had an exception and poll_thread_error is set
    # - the loop quit because exit_condition was set
    #
    # We synchronize the latch (from wait) and we set exit_condition in
    # the synchronization section to ensure that {#wait} can check for
    # exit_condition being set and set the latch only if it is not
    last_state = nil
    while !exit_condition.set?
        if latch = sync_latch.value
            latch.count_down
        end

        time = Time.now
        state = orocos_task.rtt_state
        if (state != last_state) || !current_state.value
            current_state.reset(state)
            state_queue.push(state)
            last_state = state
        end
        if latch
            latch.count_down
            sync_latch.reset(nil)
        end
        spent = (Time.now - time)
        if spent < period
            exit_condition.wait(period - spent)
        end
        run_event.wait
    end

rescue Exception => e
    poll_thread_error.reset(e)
ensure
    poll_thread_exit_sync.synchronize do
        exit_condition.set
        if latch = sync_latch.value
            latch.count_down
            latch.count_down
        end
    end
end

read() click to toggle source

Read either a new or the last read state

# File lib/syskit/remote_state_getter.rb, line 131
def read
    read_new || last_read_state
end

read_new() click to toggle source

Read a new state change

# File lib/syskit/remote_state_getter.rb, line 136
def read_new
    if new_state = state_queue.pop(true)
        @last_read_state = new_state
    end
rescue ThreadError
end

ready?() click to toggle source

Whether the state reader has read at least one state

# File lib/syskit/remote_state_getter.rb, line 126
def ready?
    last_read_state || (state_queue.size > 0)
end

resume() click to toggle source

Resume polling after a {#pause}

It is safe to call even if the polling is currently active

# File lib/syskit/remote_state_getter.rb, line 162
def resume
    run_event.set
end

wait() click to toggle source

Wait for the current state to be read and return it

# File lib/syskit/remote_state_getter.rb, line 102
def wait
    latch = poll_thread_exit_sync.synchronize do
        if !run_event.set?
            raise ThreadError, "#{self} is paused, cannot call #wait"
        elsif error = poll_thread_error.value
            raise error, "#{self}'s poll thread quit with #{error.message}, cannot call #wait", (error.backtrace + caller)
        elsif exit_condition.set?
            raise ThreadError, "#{self} is quitting, cannot call #wait"
        end
        sync_latch.reset(latch = Concurrent::CountDownLatch.new(2))
        latch
    end

    latch.wait
    if error = poll_thread_error.value
        raise error, "#{self}'s poll thread quit with #{error.message} during #wait", (error.backtrace + caller)
    elsif exit_condition.set?
        raise ThreadError, "#{self}#disconnect called within #wait"
    else
        current_state.value
    end
end