event based model - best way to implement?

[snacktime]

------=_Part_7666_9245421.1129017122016
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable
Content-Disposition: inline

I am currently using perl's POE to communicate with the asterisk manager
interface, but I would like a ruby implementation. A server will connect to
asterisk and read events as they become available. The webserver will
connnect periodically to retrieve any available events and to optionally
send more requests.

My first thought was to use a simple select loop in the server combined wit=
h
observers, and have the webserver communicate via drb. I could use http for
the server also to make it easy for ajax calls to communicate directly to
the server (proxied via apache).

The asterisk manager interface sends events to the connected client (my
server) pretty much ad hoc whenever asterisk has something to report. In
other words requests and responses are not necessarily sequential.

Does a simple select loop sound like the way to go here?

Chris

------=_Part_7666_9245421.1129017122016--

 

[Robert Klemme]

I am currently using perl's POE to communicate with the asterisk
manager interface, but I would like a ruby implementation.

Concluding from a quick glance at POE all it does is already built into
Ruby (Threads, Queue, synchronization).

A server
will connect to asterisk and read events as they become available.
The webserver will connnect periodically to retrieve any available
events and to optionally send more requests.

My first thought was to use a simple select loop in the server
combined with observers, and have the webserver communicate via drb.
I could use http for the server also to make it easy for ajax calls
to communicate directly to the server (proxied via apache).

The asterisk manager interface sends events to the connected client
(my server) pretty much ad hoc whenever asterisk has something to
report. In other words requests and responses are not necessarily
sequential.

Does a simple select loop sound like the way to go here?

You architecture is not fully clear to me, but given the small about of
sources and sinks I'd use a thread per source and either do the processing
directly in that thread or push tasks down a queue. IMHO this is far
easier in this case than writing a select loop that does the multiplexing.
Does that sound reasonable to you?

Kind regards

robert

 

[Ara.T.Howard]

I am currently using perl's POE to communicate with the asterisk manager
interface, but I would like a ruby implementation. A server will connect to
asterisk and read events as they become available. The webserver will
connnect periodically to retrieve any available events and to optionally
send more requests.

My first thought was to use a simple select loop in the server combined with
observers, and have the webserver communicate via drb. I could use http for
the server also to make it easy for ajax calls to communicate directly to
the server (proxied via apache).

The asterisk manager interface sends events to the connected client (my
server) pretty much ad hoc whenever asterisk has something to report. In
other words requests and responses are not necessarily sequential.

Does a simple select loop sound like the way to go here?

are you on windows?

-a
--
===============================================================================
| email :: ara [dot] t [dot] howard [at] noaa [dot] gov
| phone :: 303.497.6469
| Your life dwells amoung the causes of death
| Like a lamp standing in a strong breeze. --Nagarjuna
===============================================================================

 

[Paul Brannan]

Have you considered using the reactor pattern?

http://www.cs.wustl.edu/~schmidt/PDF/reactor-siemens.pdf
http://en.wikipedia.org/wiki/Reactor_Pattern

A couple of ruby implementations:

http://www.deveiate.org/projects/IO-Reactor/
http://rubystuff.org/treasures/RubyTreasures-0.5/lib/reactor/

Paul

 

[snacktime]

------=_Part_16171_24340037.1129048397466
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable
Content-Disposition: inline

 

[snacktime]

------=_Part_16395_27835985.1129048917386
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable
Content-Disposition: inline

are you on windows?

No, Freebsd.

------=_Part_16395_27835985.1129048917386--

 

[Ara.T.Howard]

No, Freebsd.

you're options are open then. select works, so does io/wait.

-a
--
===============================================================================
| email :: ara [dot] t [dot] howard [at] noaa [dot] gov
| phone :: 303.497.6469
| Your life dwells amoung the causes of death
| Like a lamp standing in a strong breeze. --Nagarjuna
===============================================================================

 

[Robert Klemme]

It was late and my brain was tired.. My question really involved around t= he
best way to constantly read from a socket in ruby and at the same time be
able to detect when a client connects with data that needs to be written = to
the socket.

Here's a typical example which uses a reader thread per client

require 'socket'
port =3D (ARGV[0] || 80).to_i
server =3D TCPServer.new('localhost', port)
while (session =3D server.accept)
Thread.new(session) do |sess|
puts "Request: #{sess.gets}"
sess.print "HTTP/1.1 200/OK\r\nContent-type: text/html\r\n\r\n"
sess.print "<html><body><h1>#{Time.now}</h1></body></html>\r\n"
# other stuff
sess.close
end
end

Derived from
http://www.ruby-doc.org/docs/ProgrammingRuby/html/lib_network.html

After looking around some more it looks like in the thread that holds the
connection to asterisk a simple loop combined with io/wait would work. An= y
reason to use io/wait over select in a case like this?

As I said, IMHO it's simpler and cleaner. Otherwise you'll be
implementing the multiplexing loop with select in Ruby which is
implemented in the interpreter in C anyway. Using select seems only
appropriate to me if the number of descriptors becomes very large.

Kind regards

robert

 

[Paul Brannan]

Here's a typical example which uses a reader thread per client

.require 'socket'
.port = (ARGV[0] || 80).to_i
.server = TCPServer.new('localhost', port)
.while (session = server.accept)
. Thread.new(session) do |sess|
. puts "Request: #{sess.gets}"
. sess.print "HTTP/1.1 200/OK\r\nContent-type: text/html\r\n\r\n"
. sess.print "<html><body><h1>#{Time.now}</h1></body></html>\r\n"
. # other stuff
. sess.close
. end
.end

You can use gserver for this, too:

require 'gserver'
class MyServer < GServer
def serve(io)
puts "Request: #{sess.gets}"
sess.print "HTTP/1.1 200/OK\r\nContent-type: text/html\r\n\r\n"
sess.print "<html><body><h1>#{Time.now}</h1></body></html>\r\n"
# other stuff
sess.close
end
end
port = (ARGV[0] || 80).to_i
server = MyServer.new(port)
server.start
server.join

Paul

 

[snacktime]

------=_Part_4260_7882869.1129139900491
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable
Content-Disposition: inline

I've gotten a ways on this project and now have another hurdle to cross.

Right now I have a server that connects to the asterisk manager interface (=
a
simple tcp line based protocol) and stays connected, acting as a kind of
proxy for connecting clients instead of making a new connection to asterisk
for each request.

One thread constantly reads events from asterisk. Each event is stuck into =
a
hash and the hash is pushed onto an array. Using the array like this can
change, it just happens to be what I've been using so far.

Clients connect to the server via drb with a request which is sent to
asterisk. The client then waits until a response is available, or until a
timeout is reached. Each client request is tagged with a unique id when it
is sent to asterisk, and asterisk returns that unique id in the response.



So basically the (abbreviated) code structure is like this, with
some_request_method being the method that is called from the drb client.
What I'm not sure about is how some_request_method will be able to know whe=
n
the response is available, or actually what would be the right way to do
this. some_request_method should block until a response is available.

class Asterisk
@events =3D []
@events.extend(MonitorMixin)
@events_pending =3D @events.new_cond

def run
reader =3D Thread.new
do loop
# reading from socket. Each event goes into a hash which is pushed onto
@events.
end
end
end

def writesock(hash)
## synchronized write to socket connected to asterisk
end

def some_request_method(hash)
writesock(hash)
# Wait for a response, which will be a hash
return response
end

end

------=_Part_4260_7882869.1129139900491--

 

[John Carter]

Just some general observations that may help....

* Threads, select, poll, reactor, state machines are all isomorphic. ie.
You can translate any solution written using any of those mechanisms
into any other. The trick is to choose which ever is easiest to code
for your particular problem.

* Threads in Ruby are implemented as a stonking great big select deep in
the C implementation of Ruby. ie. No matter what you do, at the OS
level you are doing a "select"

* On some systems "select" is implemented as a call to "poll" )

You may have observed it is _very_ easy to write ruby internal iterators,
(the each / yield paradigm), but harder (sometimes lots harder) to write a
C++'ish external iterator. (.more?, .at, .next paradigm).

Hint: You can _always_ change any solution involving an external iterator
into one involving only internal iterators, a Thread and a queue.

The last biggish chunk of code I wrote doing this sort of thing I found I
could delete a _lot_ of code and make everything lucidly straight forward
and simple by dropping the external iterator and spinning a Thread
communicating via a Queue.

Hint: By making the Queue a Queue of buffers (Arrays), the locking
performance hit is vanishingly small.

Hint: By using communicating processes, (ie. pipe & fork instead of
Thread.new) lots of obscure Threading race conditions go away.

John Carter Phone : (64)(3) 358 6639
Tait Electronics Fax : (64)(3) 359 4632
PO Box 1645 Christchurch Email : (e-mail address removed)
New Zealand

Carter's Clarification of Murphy's Law.

"Things only ever go right so that they may go more spectacularly wrong later."

From this principle, all of life and physics may be deduced.

 

[Robert Klemme]

I've gotten a ways on this project and now have another hurdle to
cross.

Right now I have a server that connects to the asterisk manager
interface (a simple tcp line based protocol) and stays connected,
acting as a kind of proxy for connecting clients instead of making a
new connection to asterisk for each request.

One thread constantly reads events from asterisk. Each event is stuck
into a hash and the hash is pushed onto an array. Using the array
like this can change, it just happens to be what I've been using so
far.

Usually one would use a Queue instead of the array because the Queue is
thread safe. You'll find one shipped in "thread". However, in this case
I'd probably use something hash like.

Clients connect to the server via drb with a request which is sent to
asterisk. The client then waits until a response is available, or
until a timeout is reached. Each client request is tagged with a
unique id when it is sent to asterisk, and asterisk returns that
unique id in the response.



So basically the (abbreviated) code structure is like this, with
some_request_method being the method that is called from the drb
client. What I'm not sure about is how some_request_method will be
able to know when the response is available, or actually what would
be the right way to do this. some_request_method should block until a
response is available.

Yep. You can achieve this by using a ConditionVariable. You can see
sample usage in my self grown sample queue implementation here
http://www.rubygarden.org/ruby?MultiThreading

I'd encapsulate this in a class. Something along the lines of this

class ResultRepository
def initialize
@results = {}
@cond = ConditionVariable.new
@mutex = Mutex.new
end

def put_result(id, result)
@mutex.synchronize do
@results[id]=result
@cond.signal
end
end

# will block
def get_result(id)
@mutex.synchronize do
@cond.wait(mutex) until @results.contains_key id
return @results.delete id
end
end
end

Kind regards

robert