/
parallel_db.txt
114 lines (81 loc) · 4.33 KB
/
parallel_db.txt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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
103
104
105
106
107
108
109
110
111
112
113
114
.. _parallel_db:
=======================
IPython's Task Database
=======================
The IPython Hub stores all task requests and results in a database. Currently supported backends
are: MongoDB, SQLite (the default), and an in-memory DictDB. The most common use case for
this is clients requesting results for tasks they did not submit, via:
.. sourcecode:: ipython
In [1]: rc.get_result(task_id)
However, since we have this DB backend, we provide a direct query method in the :class:`client`
for users who want deeper introspection into their task history. The :meth:`db_query` method of
the Client is modeled after MongoDB queries, so if you have used MongoDB it should look
familiar. In fact, when the MongoDB backend is in use, the query is relayed directly. However,
when using other backends, the interface is emulated and only a subset of queries is possible.
.. seealso::
MongoDB query docs: http://www.mongodb.org/display/DOCS/Querying
:meth:`Client.db_query` takes a dictionary query object, with keys from the TaskRecord key list,
and values of either exact values to test, or MongoDB queries, which are dicts of The form:
``{'operator' : 'argument(s)'}``. There is also an optional `keys` argument, that specifies
which subset of keys should be retrieved. The default is to retrieve all keys excluding the
request and result buffers. :meth:`db_query` returns a list of TaskRecord dicts. Also like
MongoDB, the `msg_id` key will always be included, whether requested or not.
TaskRecord keys:
=============== =============== =============
Key Type Description
=============== =============== =============
msg_id uuid(bytes) The msg ID
header dict The request header
content dict The request content (likely empty)
buffers list(bytes) buffers containing serialized request objects
submitted datetime timestamp for time of submission (set by client)
client_uuid uuid(bytes) IDENT of client's socket
engine_uuid uuid(bytes) IDENT of engine's socket
started datetime time task began execution on engine
completed datetime time task finished execution (success or failure) on engine
resubmitted datetime time of resubmission (if applicable)
result_header dict header for result
result_content dict content for result
result_buffers list(bytes) buffers containing serialized request objects
queue bytes The name of the queue for the task ('mux' or 'task')
pyin <unused> Python input (unused)
pyout <unused> Python output (unused)
pyerr <unused> Python traceback (unused)
stdout str Stream of stdout data
stderr str Stream of stderr data
=============== =============== =============
MongoDB operators we emulate on all backends:
========== =================
Operator Python equivalent
========== =================
'$in' in
'$nin' not in
'$eq' ==
'$ne' !=
'$ge' >
'$gte' >=
'$le' <
'$lte' <=
========== =================
The DB Query is useful for two primary cases:
1. deep polling of task status or metadata
2. selecting a subset of tasks, on which to perform a later operation (e.g. wait on result, purge records, resubmit,...)
Example Queries
===============
To get all msg_ids that are not completed, only retrieving their ID and start time:
.. sourcecode:: ipython
In [1]: incomplete = rc.db_query({'complete' : None}, keys=['msg_id', 'started'])
All jobs started in the last hour by me:
.. sourcecode:: ipython
In [1]: from datetime import datetime, timedelta
In [2]: hourago = datetime.now() - timedelta(1./24)
In [3]: recent = rc.db_query({'started' : {'$gte' : hourago },
'client_uuid' : rc.session.session})
All jobs started more than an hour ago, by clients *other than me*:
.. sourcecode:: ipython
In [3]: recent = rc.db_query({'started' : {'$le' : hourago },
'client_uuid' : {'$ne' : rc.session.session}})
Result headers for all jobs on engine 3 or 4:
.. sourcecode:: ipython
In [1]: uuids = map(rc._engines.get, (3,4))
In [2]: hist34 = rc.db_query({'engine_uuid' : {'$in' : uuids }, keys='result_header')