Module: IOPromise::Dalli::AsyncServer

Defined in:
lib/iopromise/dalli/patch_dalli.rb

Instance Method Summary collapse

Instance Method Details

#async? ⇒ Boolean

Returns:

  • (Boolean) 


46
47
48
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 46

def async?
  @async
end

#async_io_ready(readable, writable) ⇒ Object 



76
77
78
79
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 76

def async_io_ready(readable, writable)
  async_sock_write_nonblock if writable
  async_sock_read_nonblock if readable
end

#async_reset ⇒ Object 



66
67
68
69
70
71
72
73
74
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 66

def async_reset
  @write_buffer.clear
  @write_offset = 0

  @read_buffer.clear
  @read_offset = 0

  @executor_pool.close_socket if defined? @executor_pool
end

#close ⇒ Object 



50
51
52
53
54
55
56
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 50

def close
  if async?
    async_reset
  end

  super
end

#connect ⇒ Object 



58
59
60
61
62
63
64
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 58

def connect
  super

  if async?
    @executor_pool.connected_socket(@sock)
  end
end

#execute_continue ⇒ Object

called by ExecutorPool to continue processing for this server



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 82

def execute_continue
  timeout = @options[:socket_timeout]
  @pending_ops.select! do |key, op|
    if op.timeout?
      op.reject(Timeout::Error.new)
      next false # this op is done
    end

    # let all pending operations know that they are seeing the
    # select loop. this starts the timer for the operation, because
    # it guarantees we're now working on it.
    # this is more accurate than starting the timer when we buffer
    # the write.
    op.in_select_loop

    remaining = op.timeout_remaining
    timeout = remaining if remaining < timeout

    true # keep
  end

  @executor_pool.select_timeout = timeout
  @executor_pool.set_interest(:r, !@pending_ops.empty?)
end

#initialize(attribs, options = {}) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/iopromise/dalli/patch_dalli.rb', line 29

def initialize(attribs, options = {})
  @async = options.delete(:iopromise_async) == true

  if @async
    @write_buffer = +""
    @read_buffer = +""
    async_reset

    @next_opaque_id = 0
    @pending_ops = {}

    @executor_pool = DalliExecutorPool.for(self)
  end

  super
end