To Queue or not to Queue?

[Jeff Eberl]

I have a python script which comes up with a measurement every 30 seconds or so. It keeps a history of these measurements with timestamps.

I also am using bottle to serve a web interface to these measurements, and plot them in real time. bottle injects the history of measurements into the initial load of the web page, and the web page's javascript opens a connection to receive a Server Sent Event each time there is a new measurement, and then plots it (using flot).

What I've done is:

def measureForever():
  while True:
    data = getMeasurement()
    ??? = data
    gevent.sleep(30)

@web.route('/measurements')
def measurement_sse():
  while True:
    data = ???
    yield formatData(data)
    gevent.sleep(0)

If I use a gevent.queue for the ???, then there will be old data in the queue when a web page first requests the data. I tried using AsyncResult, but then there was always data present.

Should I use a queue, and just constantly call get in the measureForever function before I call put? Is there a type of queue (maybe priority) that will push other data out of the way?

Thanks,
Jeff

[Matt Billenstein]

What about something like:

https://gist.github.com/mattbillenstein/7baa1d0d006cf6981942

I use Event()s to synchronize the writer to the readers - and a deque to store
the data. When you render a page, you render in the deque, then stream new
data.

Is there a better way to synchronize things than using two events?

A single queue doesn't handle fanning out the data to multiple readers -
multiple queues (one per reader) seems awkward.

m

> --
> You received this message because you are subscribed to the Google Groups
> "gevent: coroutine-based Python network library" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to gevent+un...@googlegroups.com.
> For more options, visit https://groups.google.com/d/optout.


--
Matt Billenstein
ma...@vazor.com
http://www.vazor.com/

[Damjan]

What I ended up in a similar demo was,
each WSGI worker greenlet would "register" with the "broadcaster", which actually creates a Queue per greenlet, and then broadcast puts data in each Queue (for each greenlet). Don't forget to reregister the Queue when the connection ends. Pseudo code

def sse():
    q = Broadcaster.register()
    for message in q:
        yield 'event: %(event)s\ndata: %(data)s\nid: %(id)s\n\n' % message
    Broadcaster.deregister(q)


def measure():
    while True:
        data = get_data()
        Broadcaster.broadcast(data)