Intern all the things
This commit is contained in:
parent
f96526ffc2
commit
acdfef7b14
8 changed files with 97 additions and 45 deletions
|
@ -14,7 +14,7 @@
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
|
||||||
from synapse.util.frozenutils import freeze
|
from synapse.util.frozenutils import freeze
|
||||||
from synapse.util.caches import intern_string
|
from synapse.util.caches import intern_dict
|
||||||
|
|
||||||
|
|
||||||
# Whether we should use frozen_dict in FrozenEvent. Using frozen_dicts prevents
|
# Whether we should use frozen_dict in FrozenEvent. Using frozen_dicts prevents
|
||||||
|
@ -143,14 +143,7 @@ class FrozenEvent(EventBase):
|
||||||
|
|
||||||
# We intern these strings because they turn up a lot (especially when
|
# We intern these strings because they turn up a lot (especially when
|
||||||
# caching).
|
# caching).
|
||||||
event_dict["type"] = intern_string(event_dict["type"])
|
event_dict = intern_dict(event_dict)
|
||||||
if "state_key" in event_dict:
|
|
||||||
event_dict["state_key"] = intern_string(event_dict["state_key"])
|
|
||||||
if "sender" in event_dict:
|
|
||||||
event_dict["sender"] = intern_string(event_dict["sender"])
|
|
||||||
|
|
||||||
event_dict["event_id"] = intern(event_dict["event_id"].encode('ascii'))
|
|
||||||
event_dict["room_id"] = intern(event_dict["room_id"].encode('ascii'))
|
|
||||||
|
|
||||||
if USE_FROZEN_DICTS:
|
if USE_FROZEN_DICTS:
|
||||||
frozen_dict = freeze(event_dict)
|
frozen_dict = freeze(event_dict)
|
||||||
|
|
|
@ -418,6 +418,7 @@ class FederationClient(FederationBase):
|
||||||
"Failed to make_%s via %s: %s",
|
"Failed to make_%s via %s: %s",
|
||||||
membership, destination, e.message
|
membership, destination, e.message
|
||||||
)
|
)
|
||||||
|
raise
|
||||||
|
|
||||||
raise RuntimeError("Failed to send to any server.")
|
raise RuntimeError("Failed to send to any server.")
|
||||||
|
|
||||||
|
|
|
@ -175,7 +175,7 @@ class BaseFederationServlet(object):
|
||||||
|
|
||||||
|
|
||||||
class FederationSendServlet(BaseFederationServlet):
|
class FederationSendServlet(BaseFederationServlet):
|
||||||
PATH = "/send/([^/]*)/"
|
PATH = "/send/(?P<transaction_id>[^/]*)/"
|
||||||
|
|
||||||
def __init__(self, handler, server_name, **kwargs):
|
def __init__(self, handler, server_name, **kwargs):
|
||||||
super(FederationSendServlet, self).__init__(
|
super(FederationSendServlet, self).__init__(
|
||||||
|
@ -250,7 +250,7 @@ class FederationPullServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationEventServlet(BaseFederationServlet):
|
class FederationEventServlet(BaseFederationServlet):
|
||||||
PATH = "/event/([^/]*)/"
|
PATH = "/event/(?P<event_id>[^/]*)/"
|
||||||
|
|
||||||
# This is when someone asks for a data item for a given server data_id pair.
|
# This is when someone asks for a data item for a given server data_id pair.
|
||||||
def on_GET(self, origin, content, query, event_id):
|
def on_GET(self, origin, content, query, event_id):
|
||||||
|
@ -258,7 +258,7 @@ class FederationEventServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationStateServlet(BaseFederationServlet):
|
class FederationStateServlet(BaseFederationServlet):
|
||||||
PATH = "/state/([^/]*)/"
|
PATH = "/state/(?P<context>[^/]*)/"
|
||||||
|
|
||||||
# This is when someone asks for all data for a given context.
|
# This is when someone asks for all data for a given context.
|
||||||
def on_GET(self, origin, content, query, context):
|
def on_GET(self, origin, content, query, context):
|
||||||
|
@ -270,7 +270,7 @@ class FederationStateServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationBackfillServlet(BaseFederationServlet):
|
class FederationBackfillServlet(BaseFederationServlet):
|
||||||
PATH = "/backfill/([^/]*)/"
|
PATH = "/backfill/(?P<context>[^/]*)/"
|
||||||
|
|
||||||
def on_GET(self, origin, content, query, context):
|
def on_GET(self, origin, content, query, context):
|
||||||
versions = query["v"]
|
versions = query["v"]
|
||||||
|
@ -285,7 +285,7 @@ class FederationBackfillServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationQueryServlet(BaseFederationServlet):
|
class FederationQueryServlet(BaseFederationServlet):
|
||||||
PATH = "/query/([^/]*)"
|
PATH = "/query/(?P<query_type>[^/]*)"
|
||||||
|
|
||||||
# This is when we receive a server-server Query
|
# This is when we receive a server-server Query
|
||||||
def on_GET(self, origin, content, query, query_type):
|
def on_GET(self, origin, content, query, query_type):
|
||||||
|
@ -296,7 +296,7 @@ class FederationQueryServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationMakeJoinServlet(BaseFederationServlet):
|
class FederationMakeJoinServlet(BaseFederationServlet):
|
||||||
PATH = "/make_join/([^/]*)/([^/]*)"
|
PATH = "/make_join/(?P<context>[^/]*)/(?P<user_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_GET(self, origin, content, query, context, user_id):
|
def on_GET(self, origin, content, query, context, user_id):
|
||||||
|
@ -305,7 +305,7 @@ class FederationMakeJoinServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationMakeLeaveServlet(BaseFederationServlet):
|
class FederationMakeLeaveServlet(BaseFederationServlet):
|
||||||
PATH = "/make_leave/([^/]*)/([^/]*)"
|
PATH = "/make_leave/(?P<context>[^/]*)/(?P<user_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_GET(self, origin, content, query, context, user_id):
|
def on_GET(self, origin, content, query, context, user_id):
|
||||||
|
@ -314,7 +314,7 @@ class FederationMakeLeaveServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationSendLeaveServlet(BaseFederationServlet):
|
class FederationSendLeaveServlet(BaseFederationServlet):
|
||||||
PATH = "/send_leave/([^/]*)/([^/]*)"
|
PATH = "/send_leave/(?P<room_id>[^/]*)/(?P<txid>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_PUT(self, origin, content, query, room_id, txid):
|
def on_PUT(self, origin, content, query, room_id, txid):
|
||||||
|
@ -323,14 +323,14 @@ class FederationSendLeaveServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationEventAuthServlet(BaseFederationServlet):
|
class FederationEventAuthServlet(BaseFederationServlet):
|
||||||
PATH = "/event_auth/([^/]*)/([^/]*)"
|
PATH = "/event_auth(?P<context>[^/]*)/(?P<event_id>[^/]*)"
|
||||||
|
|
||||||
def on_GET(self, origin, content, query, context, event_id):
|
def on_GET(self, origin, content, query, context, event_id):
|
||||||
return self.handler.on_event_auth(origin, context, event_id)
|
return self.handler.on_event_auth(origin, context, event_id)
|
||||||
|
|
||||||
|
|
||||||
class FederationSendJoinServlet(BaseFederationServlet):
|
class FederationSendJoinServlet(BaseFederationServlet):
|
||||||
PATH = "/send_join/([^/]*)/([^/]*)"
|
PATH = "/send_join/(?P<context>[^/]*)/(?P<event_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_PUT(self, origin, content, query, context, event_id):
|
def on_PUT(self, origin, content, query, context, event_id):
|
||||||
|
@ -341,7 +341,7 @@ class FederationSendJoinServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationInviteServlet(BaseFederationServlet):
|
class FederationInviteServlet(BaseFederationServlet):
|
||||||
PATH = "/invite/([^/]*)/([^/]*)"
|
PATH = "/invite/(?P<context>[^/]*)/(?P<event_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_PUT(self, origin, content, query, context, event_id):
|
def on_PUT(self, origin, content, query, context, event_id):
|
||||||
|
@ -352,7 +352,7 @@ class FederationInviteServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationThirdPartyInviteExchangeServlet(BaseFederationServlet):
|
class FederationThirdPartyInviteExchangeServlet(BaseFederationServlet):
|
||||||
PATH = "/exchange_third_party_invite/([^/]*)"
|
PATH = "/exchange_third_party_invite/(?P<room_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_PUT(self, origin, content, query, room_id):
|
def on_PUT(self, origin, content, query, room_id):
|
||||||
|
@ -381,7 +381,7 @@ class FederationClientKeysClaimServlet(BaseFederationServlet):
|
||||||
|
|
||||||
|
|
||||||
class FederationQueryAuthServlet(BaseFederationServlet):
|
class FederationQueryAuthServlet(BaseFederationServlet):
|
||||||
PATH = "/query_auth/([^/]*)/([^/]*)"
|
PATH = "/query_auth/(?P<context>[^/]*)/(?P<event_id>[^/]*)"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_POST(self, origin, content, query, context, event_id):
|
def on_POST(self, origin, content, query, context, event_id):
|
||||||
|
@ -394,7 +394,7 @@ class FederationQueryAuthServlet(BaseFederationServlet):
|
||||||
|
|
||||||
class FederationGetMissingEventsServlet(BaseFederationServlet):
|
class FederationGetMissingEventsServlet(BaseFederationServlet):
|
||||||
# TODO(paul): Why does this path alone end with "/?" optional?
|
# TODO(paul): Why does this path alone end with "/?" optional?
|
||||||
PATH = "/get_missing_events/([^/]*)/?"
|
PATH = "/get_missing_events/(?P<room_id>[^/]*)/?"
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_POST(self, origin, content, query, room_id):
|
def on_POST(self, origin, content, query, room_id):
|
||||||
|
|
|
@ -18,6 +18,7 @@ from synapse.api.errors import (
|
||||||
cs_exception, SynapseError, CodeMessageException, UnrecognizedRequestError, Codes
|
cs_exception, SynapseError, CodeMessageException, UnrecognizedRequestError, Codes
|
||||||
)
|
)
|
||||||
from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
|
from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
|
||||||
|
from synapse.util.caches import intern_dict
|
||||||
import synapse.metrics
|
import synapse.metrics
|
||||||
import synapse.events
|
import synapse.events
|
||||||
|
|
||||||
|
@ -229,11 +230,12 @@ class JsonResource(HttpServer, resource.Resource):
|
||||||
else:
|
else:
|
||||||
servlet_classname = "%r" % callback
|
servlet_classname = "%r" % callback
|
||||||
|
|
||||||
args = [
|
kwargs = intern_dict({
|
||||||
urllib.unquote(u).decode("UTF-8") if u else u for u in m.groups()
|
name: urllib.unquote(value).decode("UTF-8") if value else value
|
||||||
]
|
for name, value in m.groupdict().items()
|
||||||
|
})
|
||||||
|
|
||||||
callback_return = yield callback(request, *args)
|
callback_return = yield callback(request, **kwargs)
|
||||||
if callback_return is not None:
|
if callback_return is not None:
|
||||||
code, response = callback_return
|
code, response = callback_return
|
||||||
self._send_response(request, code, response)
|
self._send_response(request, code, response)
|
||||||
|
|
|
@ -18,6 +18,7 @@ from synapse.api.errors import StoreError
|
||||||
from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
|
from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
|
||||||
from synapse.util.caches.dictionary_cache import DictionaryCache
|
from synapse.util.caches.dictionary_cache import DictionaryCache
|
||||||
from synapse.util.caches.descriptors import Cache
|
from synapse.util.caches.descriptors import Cache
|
||||||
|
from synapse.util.caches import intern_dict
|
||||||
import synapse.metrics
|
import synapse.metrics
|
||||||
|
|
||||||
|
|
||||||
|
@ -350,7 +351,7 @@ class SQLBaseStore(object):
|
||||||
"""
|
"""
|
||||||
col_headers = list(column[0] for column in cursor.description)
|
col_headers = list(column[0] for column in cursor.description)
|
||||||
results = list(
|
results = list(
|
||||||
dict(zip(col_headers, row)) for row in cursor.fetchall()
|
intern_dict(dict(zip(col_headers, row))) for row in cursor.fetchall()
|
||||||
)
|
)
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
|
@ -62,18 +62,17 @@ class ReceiptsStore(SQLBaseStore):
|
||||||
|
|
||||||
@cachedInlineCallbacks(num_args=2)
|
@cachedInlineCallbacks(num_args=2)
|
||||||
def get_receipts_for_user(self, user_id, receipt_type):
|
def get_receipts_for_user(self, user_id, receipt_type):
|
||||||
def f(txn):
|
rows = yield self._simple_select_list(
|
||||||
sql = (
|
table="receipts_linearized",
|
||||||
"SELECT room_id,event_id "
|
keyvalues={
|
||||||
"FROM receipts_linearized "
|
"user_id": user_id,
|
||||||
"WHERE user_id = ? AND receipt_type = ? "
|
"receipt_type": receipt_type,
|
||||||
)
|
},
|
||||||
txn.execute(sql, (user_id, receipt_type))
|
retcols=("room_id", "event_id"),
|
||||||
return txn.fetchall()
|
desc="get_receipts_for_user",
|
||||||
|
)
|
||||||
|
|
||||||
defer.returnValue(dict(
|
defer.returnValue({row["room_id"]: row["event_id"] for row in rows})
|
||||||
(yield self.runInteraction("get_receipts_for_user", f))
|
|
||||||
))
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def get_linearized_receipts_for_rooms(self, room_ids, to_key, from_key=None):
|
def get_linearized_receipts_for_rooms(self, room_ids, to_key, from_key=None):
|
||||||
|
|
|
@ -156,9 +156,7 @@ class StateStore(SQLBaseStore):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def get_current_state_for_key(self, room_id, event_type, state_key):
|
def get_current_state_for_key(self, room_id, event_type, state_key):
|
||||||
event_ids = yield self._get_current_state_for_key(
|
event_ids = yield self._get_current_state_for_key(room_id, event_type, state_key)
|
||||||
room_id, intern_string(event_type), intern_string(state_key)
|
|
||||||
)
|
|
||||||
events = yield self._get_events(event_ids, get_prev_content=False)
|
events = yield self._get_events(event_ids, get_prev_content=False)
|
||||||
defer.returnValue(events)
|
defer.returnValue(events)
|
||||||
|
|
||||||
|
@ -205,7 +203,7 @@ class StateStore(SQLBaseStore):
|
||||||
|
|
||||||
results = {}
|
results = {}
|
||||||
for row in rows:
|
for row in rows:
|
||||||
key = (intern_string(row["type"]), intern_string(row["state_key"]))
|
key = (row["type"], row["state_key"])
|
||||||
results.setdefault(row["state_group"], {})[key] = row["event_id"]
|
results.setdefault(row["state_group"], {})[key] = row["event_id"]
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
@ -286,7 +284,9 @@ class StateStore(SQLBaseStore):
|
||||||
desc="_get_state_group_for_events",
|
desc="_get_state_group_for_events",
|
||||||
)
|
)
|
||||||
|
|
||||||
defer.returnValue({row["event_id"]: row["state_group"] for row in rows})
|
defer.returnValue({
|
||||||
|
intern(row["event_id"].encode('ascii')): row["state_group"] for row in rows
|
||||||
|
})
|
||||||
|
|
||||||
def _get_some_state_from_cache(self, group, types):
|
def _get_some_state_from_cache(self, group, types):
|
||||||
"""Checks if group is in cache. See `_get_state_for_groups`
|
"""Checks if group is in cache. See `_get_state_for_groups`
|
||||||
|
|
|
@ -15,6 +15,9 @@
|
||||||
|
|
||||||
import synapse.metrics
|
import synapse.metrics
|
||||||
from lrucache import LruCache
|
from lrucache import LruCache
|
||||||
|
import os
|
||||||
|
|
||||||
|
CACHE_SIZE_FACTOR = float(os.environ.get("SYNAPSE_CACHE_FACTOR", 0.1))
|
||||||
|
|
||||||
DEBUG_CACHES = False
|
DEBUG_CACHES = False
|
||||||
|
|
||||||
|
@ -27,9 +30,62 @@ cache_counter = metrics.register_cache(
|
||||||
labels=["name"],
|
labels=["name"],
|
||||||
)
|
)
|
||||||
|
|
||||||
_string_cache = LruCache(5000)
|
_string_cache = LruCache(int(5000 * CACHE_SIZE_FACTOR))
|
||||||
caches_by_name["string_cache"] = _string_cache
|
caches_by_name["string_cache"] = _string_cache
|
||||||
|
|
||||||
|
|
||||||
|
KNOWN_KEYS = {
|
||||||
|
key: key for key in
|
||||||
|
(
|
||||||
|
"auth_events",
|
||||||
|
"content",
|
||||||
|
"depth",
|
||||||
|
"event_id",
|
||||||
|
"hashes",
|
||||||
|
"origin",
|
||||||
|
"origin_server_ts",
|
||||||
|
"prev_events",
|
||||||
|
"room_id",
|
||||||
|
"sender",
|
||||||
|
"signatures",
|
||||||
|
"state_key",
|
||||||
|
"type",
|
||||||
|
"unsigned",
|
||||||
|
"user_id",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def intern_string(string):
|
def intern_string(string):
|
||||||
|
"""Takes a (potentially) unicode string and interns using custom cache
|
||||||
|
"""
|
||||||
return _string_cache.setdefault(string, string)
|
return _string_cache.setdefault(string, string)
|
||||||
|
|
||||||
|
|
||||||
|
def intern_dict(dictionary):
|
||||||
|
"""Takes a dictionary and interns well known keys and their values
|
||||||
|
"""
|
||||||
|
return _intern_known_values({
|
||||||
|
_intern_key(key): value for key, value in dictionary.items()
|
||||||
|
})
|
||||||
|
|
||||||
|
|
||||||
|
def _intern_known_values(dictionary):
|
||||||
|
intern_str_keys = ("event_id", "room_id")
|
||||||
|
intern_unicode_keys = ("sender", "user_id", "type", "state_key")
|
||||||
|
|
||||||
|
for key in intern_str_keys:
|
||||||
|
val = dictionary.get(key, None)
|
||||||
|
if val is not None:
|
||||||
|
dictionary[key] = intern(val.encode('ascii'))
|
||||||
|
|
||||||
|
for key in intern_unicode_keys:
|
||||||
|
val = dictionary.get(key, None)
|
||||||
|
if val is not None:
|
||||||
|
dictionary[key] = intern_string(val)
|
||||||
|
|
||||||
|
return dictionary
|
||||||
|
|
||||||
|
|
||||||
|
def _intern_key(key):
|
||||||
|
return KNOWN_KEYS.get(key, key)
|
||||||
|
|
Loading…
Reference in a new issue