Class: ZMQPoll::ZMQPoll

Inherits:
Object
  • Object
show all
Defined in:
lib/log-courier/zmq_qpoll.rb

Instance Method Summary collapse

Constructor Details

#initialize(context, logger = nil) ⇒ ZMQPoll

Returns a new instance of ZMQPoll.



31
32
33
34
35
36
37
38
39
 
# File 'lib/log-courier/zmq_qpoll.rb', line 31

def initialize(context, logger=nil)
	@logger = logger
	@context = context
	@poller = ZMQ::Poller.new
	@sockets = []
	@socket_to_socket = []
	@handlers = {}
	@queues = {}
end

Instance Method Details

#close_socket_to_socket(socket) ⇒ Object 



126
127
128
129
130
131
132
 
# File 'lib/log-courier/zmq_qpoll.rb', line 126

def close_socket_to_socket(socket)
	return if !@socket_to_socket.include?(socket)
	state = @socket_to_socket[socket]
	@socket_to_socket.delete socket
	_close_socket_to_socket(state)
	return
end

#create_socket_to_socket(socket) ⇒ Object 



120
121
122
123
124
 
# File 'lib/log-courier/zmq_qpoll.rb', line 120

def create_socket_to_socket(socket)
	state = _create_socket_to_socket(socket)
	@socket_to_socket[state[:sender]] = state
	state[:sender]
end

#deregister_queue(queue) ⇒ Object 



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/log-courier/zmq_qpoll.rb', line 104

def deregister_queue(queue)
	return if !@queues.key?(queue)

	# Push nil so if we're idle we jump into action and exit
	# But also set shutdown to try so if we're mid-send and timeout, we exit
	@queues[queue][:mutex].synchronize do
		queue.push nil
		@queues[queue][:shutdown] = true
	end
	@queues[queue][:thread].join

	_close_socket_to_socket @queues[queue][:state]
	@queues.delete queue
	return
end

#deregister_socket(socket) ⇒ Object 



69
70
71
72
73
74
 
# File 'lib/log-courier/zmq_qpoll.rb', line 69

def deregister_socket(socket)
	return if @handlers.key?(socket)

	@poller.delete socket
	return
end

#poll(timeout) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/log-courier/zmq_qpoll.rb', line 134

def poll(timeout)
	if @poller.size == 0
		fail ZMQError, 'poll run called with zero socket/queues'
	end

	rc = @poller.poll(timeout)
	unless ZMQ::Util.resultcode_ok?(rc)
		# If we get ETERM or ENOTSOCK - we may have hit JRuby bug where finalisers are called too early
		# We throw it so the caller knows NOT to retry and hit a bad loop
		# The downside is - this Jruby bug will cause an inevitable deadlock
		# The finaliser for the context will have run before finalisers for sockets (random order?)
		# The resulting ZMQ context termination will block, waiting for sockets to be closed
		# However, when closing a socket, ffi-rzmq will attempt to remove it's finaliser
		# This removal will deadlock as it attempts to lock the finaliser list, which is locked by the
		# thread that is running the ZMQ context termination, which is blocked...
		# TODO: Raise issue in JRuby github and try to track down why finalisers run while code is running
		# if the exit! call is made
		fail ZMQTerm, 'poll error: ' + ZMQ::Util.error_string if ZMQ::Util.errno == ZMQ::ETERM || ZMQ::Util.errno == ZMQ::ENOTSOCK
		fail ZMQError, 'poll error: ' + ZMQ::Util.error_string
	end

	return if rc == 0

	ready = (@poller.readables|@poller.writables)

	ready.each do |socket|
		if @handlers.key?(socket)
			__send__ @handlers[socket][:callback], @handlers[socket]
		end

		yield socket, @poller.readables.include?(socket), @poller.writables.include?(socket)
	end

	return
end

#readables ⇒ Object 



41
42
43
 
# File 'lib/log-courier/zmq_qpoll.rb', line 41

def readables
	@poller.readables
end

#register_queue_to_socket(queue, socket) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/log-courier/zmq_qpoll.rb', line 76

def register_queue_to_socket(queue, socket)
	s2s_state = _create_socket_to_socket(socket)

	state = {
		state:    s2s_state,
		mutex:    Mutex.new,
		shutdown: false,
	}

	state[:thread] = Thread.new do
		loop do
			data = queue.pop
			break if data.nil?
			begin
				send s2s_state[:sender], data
			rescue TimeoutError
				state[:mutex].synchronize do
					break if state[:shutdown]
				end
				retry
			end
		end
	end

	@queues[queue] = state
	return
end

#register_socket(socket, flags) ⇒ Object 



64
65
66
67
 
# File 'lib/log-courier/zmq_qpoll.rb', line 64

def register_socket(socket, flags)
	@poller.register socket, flags
	return
end

#shutdown ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/log-courier/zmq_qpoll.rb', line 49

def shutdown
	@queues.each_key do |queue|
		deregister_queue queue
	end

	@socket_to_socket.each do |socket|
		_close_socket_to_socket socket
	end

	@sockets.each do |socket|
		socket.close
	end
	return
end

#writables ⇒ Object 



45
46
47
 
# File 'lib/log-courier/zmq_qpoll.rb', line 45

def writables
	@poller.writables
end