-
Notifications
You must be signed in to change notification settings - Fork 18
/
tonlibjson.py
252 lines (202 loc) · 8.92 KB
/
tonlibjson.py
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
import asyncio
import functools
import json
import logging
import platform
import random
import time
import traceback
from ctypes import *
import pkg_resources
from .tl.base import TLObject
logger = logging.getLogger('ton')
class TonlibException(Exception):
pass
class TonlibNoResponse(TonlibException):
def __str__(self):
return 'tonlibjson did not respond'
class TonlibError(TonlibException):
def __init__(self, result):
self.result = result
@property
def code(self):
return self.result.get('code')
def __str__(self):
return self.result.get('message')
class LiteServerTimeout(TonlibError):
pass
class BlockNotFound(TonlibError):
pass
class BlockDeleted(TonlibError):
pass
class ExternalMessageNotAccepted(TonlibError):
pass
def parse_tonlib_error(result):
if result.get('@type') == 'error':
message = result.get('message')
if 'not in db' in message:
return BlockNotFound(result)
if "state already gc'd" in message:
return BlockDeleted(result)
if 'cannot apply external message to current state' in message:
return ExternalMessageNotAccepted(result)
if 'adnl query timeout' in message:
return LiteServerTimeout(result)
return TonlibError(result)
return None
def get_tonlib_path():
arch_name = platform.system().lower()
machine = platform.machine().lower()
if arch_name == 'linux':
lib_name = f'libtonlibjson.{machine}.so'
elif arch_name == 'darwin':
lib_name = f'libtonlibjson.{machine}.dylib'
elif arch_name == 'freebsd':
lib_name = f'libtonlibjson.{machine}.so'
elif arch_name == 'windows':
lib_name = f'tonlibjson.{machine}.dll'
else:
raise RuntimeError(f"Platform '{arch_name}({machine})' is not compatible yet. Read more at https://github.com/psylopunk/pytonlib/issues/7")
return pkg_resources.resource_filename('ton', f'distlib/{arch_name}/{lib_name}')
class TonLib:
def __init__(self, loop, ls_index, cdll_path=None, verbosity_level=0, default_timeout=None):
self.loop = loop
self.default_timeout = default_timeout
cdll_path = get_tonlib_path() if not cdll_path else cdll_path
tonlib = CDLL(cdll_path)
tonlib_json_client_create = tonlib.tonlib_client_json_create
tonlib_json_client_create.restype = c_void_p
tonlib_json_client_create.argtypes = []
try:
self._client = tonlib_json_client_create()
except Exception as ee:
raise RuntimeError(f"Failed to create tonlibjson client: {ee}")
tonlib_json_client_receive = tonlib.tonlib_client_json_receive
tonlib_json_client_receive.restype = c_char_p
tonlib_json_client_receive.argtypes = [c_void_p, c_double]
self._tonlib_json_client_receive = tonlib_json_client_receive
tonlib_json_client_send = tonlib.tonlib_client_json_send
tonlib_json_client_send.restype = None
tonlib_json_client_send.argtypes = [c_void_p, c_char_p]
self._tonlib_json_client_send = tonlib_json_client_send
tonlib_json_client_execute = tonlib.tonlib_client_json_execute
tonlib_json_client_execute.restype = c_char_p
tonlib_json_client_execute.argtypes = [c_void_p, c_char_p]
self._tonlib_json_client_execute = tonlib_json_client_execute
tonlib_json_client_destroy = tonlib.tonlib_client_json_destroy
tonlib_json_client_destroy.restype = None
tonlib_json_client_destroy.argtypes = [c_void_p]
self._tonlib_json_client_destroy = tonlib_json_client_destroy
self.futures = {}
self.ls_index = ls_index
self._state = None # None, "finished", "crashed", "stuck"
self.is_dead = False
# creating tasks
self.read_results_task = self.loop.create_task(self.read_results())
self.del_expired_futures_task = self.loop.create_task(self.del_expired_futures_loop())
def __del__(self):
try:
self._tonlib_json_client_destroy(self._client)
except Exception as ee:
logger.error(f"Exception in tonlibjson.__del__: {traceback.format_exc()}")
raise RuntimeError(f'Error in tonlibjson.__del__: {ee}')
def send(self, query):
if not self._is_working:
raise RuntimeError(f"TonLib failed with state: {self._state}")
query = json.dumps(query).encode('utf-8')
try:
self._tonlib_json_client_send(self._client, query)
except Exception as ee:
logger.error(f"Exception in tonlibjson.send: {traceback.format_exc()}")
raise RuntimeError(f'Error in tonlibjson.send: {ee}')
def receive(self, timeout=30):
result = None
try:
result = self._tonlib_json_client_receive(self._client, timeout) # time.sleep # asyncio.sleep
except Exception as ee:
logger.error(f"Exception in tonlibjson.receive: {traceback.format_exc()}")
raise RuntimeError(f'Error in tonlibjson.receive: {ee}')
if result:
result = json.loads(result.decode('utf-8'))
return result
def _execute(self, query, timeout=30):
if not self._is_working:
raise RuntimeError(f"TonLib failed with state: {self._state}")
extra_id = "%s:%s:%s" % (time.time() + timeout, self.ls_index, random.random())
query["@extra"] = extra_id
future_result = self.loop.create_future()
self.futures[extra_id] = future_result
self.loop.run_in_executor(None, lambda: self.send(query))
return future_result
async def execute(self, query, timeout=30):
logger.debug(f'SENT' + '\n' + f'{query}')
if isinstance(query, TLObject): query = query.to_json()
result = await self._execute(query, timeout=timeout)
result = TLObject.from_json(result)
logger.debug(f'RECEIVED' + '\n' + f'{result}')
if result.type == 'updateSyncState':
await asyncio.sleep(1)
return await self.execute(query, timeout=timeout)
return result
@property
def _is_working(self):
return self._state not in ('crashed', 'stuck', 'finished')
@property
def _is_closing(self):
return self._state == 'finishing'
async def close(self):
try:
self._state = 'finishing'
await self.read_results_task
await self.del_expired_futures_task
self._state = 'finished'
except Exception as ee:
logger.error(f"Exception in tonlibjson.close: {traceback.format_exc()}")
raise RuntimeError(f'Error in tonlibjson.close: {ee}')
def cancel_futures(self, cancel_all=False):
now = time.time()
to_del = []
for i in self.futures:
if float(i.split(":")[0]) <= now or cancel_all:
to_del.append(i)
logger.debug(f'Pruning {len(to_del)} tasks')
for i in to_del:
self.futures[i].set_exception(TonlibNoResponse())
self.futures.pop(i)
# tasks
async def read_results(self):
timeout = self.default_timeout or 1
receive_func = functools.partial(self.receive, 0.1)
try:
while self._is_working and not self._is_closing:
# return reading result
result = None
try:
result = await asyncio.wait_for(self.loop.run_in_executor(None, receive_func), timeout=timeout)
except asyncio.TimeoutError:
logger.critical(f"Tonlib #{self.ls_index:03d} stuck (timeout error)")
self._state = "stuck"
except:
logger.critical(f"Tonlib #{self.ls_index:03d} crashed: {traceback.format_exc()}")
self._state = "crashed"
if isinstance(result, dict) and ("@extra" in result) and (result["@extra"] in self.futures):
try:
if not self.futures[result["@extra"]].done():
tonlib_error = parse_tonlib_error(result)
if tonlib_error is not None:
self.futures[result["@extra"]].set_exception(tonlib_error)
else:
self.futures[result["@extra"]].set_result(result)
self.futures.pop(result["@extra"])
except Exception as e:
logger.error(f'Tonlib #{self.ls_index:03d} receiving result exception: {e}')
except Exception as ee:
logger.critical(f'Task read_results failed: {ee}')
async def del_expired_futures_loop(self):
try:
while self._is_working and not self._is_closing:
self.cancel_futures()
await asyncio.sleep(1)
self.cancel_futures(cancel_all=True)
except Exception as ee:
logger.critical(f'Task del_expired_futures_loop failed: {ee}')