bitly / asyncmongo Goto Github PK

I've set up a simple asyncmongo/zeromq router to act as a message queue as well as manage writes to my datastore.

what I'm finding is that the connections are getting released or reused, so it eventually hits maxconnections and stops.

Here's what I'm working with. You can pass a few messages then it stops and starts throwing TooManyConnections

#!chatforus-ve/bin/python
# -*- coding: utf-8 -*-
#
# Copyright 2011 Joseph Bowman
#
# Posting this in a Google group so I guess
# it's public domain now.
#
import asyncmongo
import zmq
from zmq.eventloop import ioloop, zmqstream
from tornado import escape
from tornado import gen
import datetime

loop = ioloop.IOLoop.instance()

db = asyncmongo.Client(pool_id='mydb', host='127.0.0.1', port=27017, maxconnections=5, dbname='chatforus')

ctx = zmq.Context()
receiver = ctx.socket(zmq.PULL)
broadcast = ctx.socket(zmq.PUSH)
receiver.bind('tcp://127.0.0.1:5555')
broadcast.bind('tcp://127.0.0.1:5556')
receiver_stream = zmqstream.ZMQStream(receiver, loop)
broadcast_stream = zmqstream.ZMQStream(broadcast, loop)

@gen.engine
def handle_message(msg):
    broadcast_stream.send(msg[0]) 
    message = escape.json_decode(msg[0]) 
    if message["type"] == "msg":
        message["ts"] = datetime.datetime.now()
        (response,), error = yield gen.Task(db.msgs.insert, message)

receiver_stream.on_recv(handle_message)

loop.start()

There seem to be no easy way to get a list of all collections in asyncmongo.

Pymongo provides the method collection_names() on db objects.

27107 should be 27017 because it is the default port for Mongo.

I copied the example as a starting place for an app and I am not willing to admit how much time I wasted trying to track down the "[Errno 61] Connection refused" issue.

Hello, asyncmongo does a very brilliant work!

Now I'm encountered a problem that troubles me. I'm using mongos as a cluster proxy, and when the rqs is up to 2000r/s, the mongos occupied all system memory and when the rqs get higher, the request get "connection reset pee'.

And I looked at the source code of connection pool asyncmongo used, and I got that when the request reached out the maxConnections, the request will failed by the driver. But if I enlarge the maxConnections, the mongos will report error like before for that mongos manipulate each request by a thread.

Why not implement the connection pool to block the request when reach the maxConnections like Spring database connection pool implemented in JAVA instead of return an error ?

Is asyncmongo going to support a db.eval() function?

Looks like curront asyncmongo doesn't support replica set connection like what PyMongo done In their documentation here. They will autoselect the master based on the host lists.

def _on_response(self, response, error):
if error:
raise tornado.web.HTTPError(500)

self.do_something()

the asyncmongo will catch tornado.web.HTTPError and display "callback failed",rather than deliver the exception to tornado.

According to the documentation, find_one should return None when the query doesn't find anything.

What actually happens is that an empty list is returned.

Setting mincached to anything above 0 borks things:

File "/home/shuchen/virtualenvs/idx_server/lib/python2.6/site-packages/asyncmongo/pool.py", line 71, in __init__
    idle = [self.dedicated_connection() for i in range(mincached)]
AttributeError: 'ConnectionPool' object has no attribute 'dedicated_connection'

The connection decrement in ConnectionPool.cache() is NOT thread safe (asyncmongo/pool.py: 129)

It probably needs to be in the lock.

http://effbot.org/pyfaq/what-kinds-of-global-value-mutation-are-thread-safe.htm
http://mail.python.org/pipermail/python-dev/2006-November/thread.html#69981
http://www.velocityreviews.com/forums/t632419-atomic-increment.html

Support wanted for SONManipulator to allow specifying transformations to be applied automatically by asyncmongo. Corresponding class in pymongo has the same name.

when tornado autoreload, asyncmongo didn't release connections

error : function takes at most 2 arguments (3 given)

update connection.py
...
from errors import ProgrammingError, IntegrityError, InterfaceError
import helpers
import pymongo.helpers as helpers # replace helpers for pymongo 3.3.1
import asyncjobs
...

File "/home/krik/LBE3/lib/python3.2/site-packages/asyncmongo/connection.py", line 82
except socket.error, error:
^
SyntaxError: invalid syntax

  File "/home/krik/LBE3/lib/python3.2/site-packages/asyncmongo/cursor.py", line 47
    return u'%s.%s' % (self.__dbname, self.__collection)
                  ^
SyntaxError: invalid syntax

  File "/home/krik/LBE3/lib/python3.2/site-packages/asyncmongo/errors.py", line 44
    return u'IntegrityError: %s code:%s' % (self.msg, self.code or '')
                                       ^
SyntaxError: invalid syntax

I was wondering if it would be at all possible to upload asyncmongo to PyPI instead of relying on externally hosted files.

Accessing collection.group results in

AttributeError: 'Cursor' object has no attribute 'group'

The latest pypi 1.0 tarball does not include the README.md file, so the setup.py fails when trying to load it for the long_description.

It seems the tests are always installing tornado=2.4.1 and tornado 3.0 and higher is not supported yet.

Is that right? I'd like to upgrade python-tornado to 3.1.1 and it seems to be not possible with the current asyncmongo.

I'm trying to maintain the order of JSON variables, for some silly reason that matters to my project.
In the PyMongo driver, I use mongo.find({},as_class=OrderedDict) to force it to return as an OrderedDict, rather than a traditional, unordered dictionary.
This is a 2.7+ (or PyPy) feature, but it's a very useful one to me.

Is there any way using asyncmongo to force it to return using an OrderedDict, using as_class, or globally changing something?

There are cases where mongodb returns a cursor_id (in _parse_response) but that cursor is not closed explicitly.

testing for this should follow the pymongo test_kill_cursors tests.

Hi,
There is a book called MongoDB for Python and the examples are available here ( http://examples.oreilly.com/0636920021513/mongodb-python-source-and-snippets.zip ). Could someone convert them to asyncmongo, please?

Thank you in advance.

Mic

Right now passing slave_ok=True to Client does nothing. Cursor.find still throws errors when connected to a slave. First step towards replSet support.

Why do third-party libraries have their own reference problems？
from errors import (Error, InterfaceError, AuthenticationError, DatabaseError, RSConnectionError
No module named 'errors'

When running findAndyModify commands with a SON object in the following context the query argument is put at the start of the SON object thus causing an error with Mongo

from bson.son import SON
from bson import ObjectId
q = SON()
q['findAndModify'] = 'collection'
q['query'] = {"_id" : ObjectId("valid-object-id")}

database.command(q, callback=callback)

The SON object is re-ordered putting the query item in front of findAndModify and thus the query fails.

Hi,

So I finally got time to look at using tailable cursor with mongodb. I've been playing with asyncmongo a bit, planning on using it to read/write within requests. It however doesn't really work with tailable cursors because there needs to be a real cursor to iterate on, and asyncmongo doesn't seem to return those? I played around with the Cursor class, and it doesn't have support for attributes like alive and methods like next.

Here's how I set up a nonblocking tailable cursor implementation with Tornado and pymongo. Maybe something like this could be included in an asyncmongo at some point?

        self.msgs_db = pymongo.Connection(host='127.0.0.1', 
                    port=27017)
        self.msgs_cursor = self.msgs_db.mydb.mycoll.find({}, tailable=True)
        self.msgs_cursor.batch_size(10)

        self.tail()

    def tail(self, count=0):
        if count > 100:
            IOLoop.instance().add_timeout(time.time() + .5, self.tail)
        if self.msgs_cursor.alive:
            try:
                logging.error(self.msgs_cursor.next())
                self.tail(count=count+1)
            except StopIteration:
                IOLoop.instance().add_timeout(time.time() + .5, self.tail)
        else:
            self.msgs_cursor = self.msgs_db.mydb.mycoll.find({}, tailable=True)

notes: mycoll needs to be a capped collection. The purpose is to be able to just stream updates to each app server as they happen, rather than repeatedly poll the database. The count is just in case messages are coming in too fast, grabbing x at a time (100 in this case, which is probably way too low) and then freeing the loop up seems to be a way to go.

Thinking about this, it might actually be better performance to just poll, and I think I might go in that direction after all. But I didn't want to just throw the code away if it's something you might be able to use and others might have a use for.

method insert doesnt returned ObjectId() and doesn't pass it to callback function, this is a normal behavior?

File ".../env/build/asyncmongo/setup.py", line 11, in

long_description=open(os.path.join(os.path.dirname(__file__),"README.md"), "r").read(),

IOError: [Errno 2] No such file or directory: '.../env/build/asyncmongo/README.md'

This error is the result of 'pip install asyncmongo'.

ubutnu 64bit, mongodb 2.0.1, DB > 3GB, objects > 3M

asyncmongo:
self.db = asyncmongo.Client(pool_id='mydb', host='127.0.0.1', port=27017, maxcached=10, maxconnections=50, dbname='crm1')
tr = yield gen.Task(self.db.obj.find, { 'type':0 }, sort=[('when',-1)], skip = 0, limit = 15)
if tr[1]['error']:
self.do_error(tr[1]['error'])
raise StopIteration
result = tr[0][0]

sync: pymongo:
result = list( self.dbs.obj.find({ 'type':0 }, skip = 0, limit = 15) )

duration for 1st query, 2nd, 3rd, 4s...
asyncmongo: 1ms, 600ms, 4000ms, 20s...
pymongo: 1ms, 1ms, 1ms, 1ms,

on linux 32bit, asyncmongo is ok.

I'm using aysncmongo on my project, and it works cool. Thanks for you brilliant work!

And now I'm encounter a problem on using the command 'count', and a little cofused.
I have a requirement for getting the count of a query, and I coded it like:

db['$cmd'].find_one({'count' : 'mycollect'}, call_back=resp_func)

But it didn't work.....
I then noticed that there is a '_is_command' parameter for 'find()'. I also noticed that in fact it is not in use, but instead passed 'False' directly to the 'self.__is_command', which will be used for creating the 'find_spec'.
I hacked the code to use '_is_command' parameter, and the code is:

db['$cmd'].find_one({'count' : 'mycollect'}, _is_command=True, call_back=resp_func)

And it works!

I think there must be some reason for disable the 'is_command' option, and I wonder why? ^^ And also is it safe for just use the 'count' command?

Thanks a lot for the work!!

I'm packaging asyncmongo for Fedora and the reviewer asked me to ping upstream and request a LICENSE be included.

Thanks!

https://bugzilla.redhat.com/show_bug.cgi?id=654544

Non-mandatory items:
* no license file shipped in source but since all source files header mention
the actual license (vanilla ASL 2.0) it should not block the acceptation of the
package. Anyway, i recommend packager asking upstream to add one and fix the
distutils PKG-INFO file (the license field says UNKNOWN).

The python implementation of the bson library works just fine, but switching to the C-compiled version results in errors when using the bson.binary.Binary object, in particular, performing a find query using a Binary results in the query silently failing by returning nothing.

pymongo (both C and python) doesn't have this issue, so I'm thinking it might be a problem with asyncmongo.

Hi,
First off, great project. Thanks for open sourcing it.
Secondly, I can't seem to get logical operators like $or or $and to work. I could always just run my own parallel searches (which is what mongo does anyways I believe) and then logical-or it on my side, but i feel like this should work.

I'm trying to connect to a mongodb instance being hosted on a remote server. The only way to connect to the host is through the following url:

username:[email protected]

i.e. root:[email protected]

asyncmongo reports back an error,

No address associated with hostname

Any idea of a fix/workaround to be able to connect?

Thanks!
Vikas

something:

index = SON()
for (key, value) in index_list:
    if not isinstance(key, basestring):
        raise TypeError("first item in each key pair must be a string")
    if value not in [ASCENDING, DESCENDING, GEO2D, {'$meta': 'textScore'}]:
        raise TypeError("second item in each key pair must be ASCENDING, "
                        "DESCENDING, or GEO2D")
    index[key] = value
return index

Looking at the example in the README, what happens if you have more than one get() method, and they need to be able to execute async MongoDB queries - but against >1 MongoDB database. With pymongo you can just do connection.db.collection.find(); what is the best way to do this with asyncmongo? Is there a way to set the dbname with a command (such as find/insert etc.)?

GLib event loop for example.

using the limit kw arg into find, can give you more results, however when it is not specified, asyncmongo defaults to 101? the code does not say anything about this.

import asyncmongo
import pymongo
import threading
import time
import tornado

def start_loop_timeout(timeout=0.05):
def kill():
time.sleep(timeout)
tornado.ioloop.IOLoop.instance().stop()
t = threading.Thread(target=kill)
t.start()
tornado.ioloop.IOLoop.instance().start()

class AsyncDBTester(object):
def init(self):
self.responses = []
self.errors = []

def async_callback(self, response, error):
    self.responses.append(response)
    self.errors.append(error)

db = asyncmongo.Client(pool_id='test', host='127.0.0.1', port=27017, dbname='test')
AsyncDBTester = AsyncDBTester
dbtester = AsyncDBTester()
db.dummy.remove(callback=dbtester.async_callback)
start_loop_timeout(0.1)
count = 0
for b in range(10):
for c in range(100):
db.dummy.insert({'a' : b * c}, callback=dbtester.async_callback)
count += 1
start_loop_timeout(0.1)

dbtester = AsyncDBTester()
db.dummy.find(callback=dbtester.async_callback, limit=1000)
start_loop_timeout(0.2)
print 'inserted', count
print "can get by specifying limit", len(dbtester.responses[0])

dbtester = AsyncDBTester()
db.dummy.find(callback=dbtester.async_callback)
start_loop_timeout(0.2)
print count
print "but why 101 here?", len(dbtester.responses[0])

blockingdb = pymongo.connection.Connection(host = '127.0.0.1', port = 27017)['test']
result = list(blockingdb.dummy.find())
print "pymongo gives me", len(result)