@ -364,147 +364,161 @@ class EventsStore(
if not events_and_contexts :
return
if backfilled :
stream_ordering_manager = self . _backfill_id_gen . get_next_mult (
len ( events_and_contexts )
)
else :
stream_ordering_manager = self . _stream_id_gen . get_next_mult (
len ( events_and_contexts )
)
with stream_ordering_manager as stream_orderings :
for ( event , context ) , stream in zip ( events_and_contexts , stream_orderings ) :
event . internal_metadata . stream_ordering = stream
chunks = [
events_and_contexts [ x : x + 100 ]
for x in range ( 0 , len ( events_and_contexts ) , 100 )
]
for chunk in chunks :
# We can't easily parallelize these since different chunks
# might contain the same event. :(
chunks = [
events_and_contexts [ x : x + 100 ]
for x in range ( 0 , len ( events_and_contexts ) , 100 )
]
# NB: Assumes that we are only persisting events for one room
# at a time.
for chunk in chunks :
# We can't easily parallelize these since different chunks
# might contain the same event. :(
# map room_id->list[event_ids] giving the new forward
# extremities in each room
new_forward_extremeties = { }
# NB: Assumes that we are only persisting events for one room
# at a time.
# map room_id->(type,state_key)->event_id tracking the full
# state in each room after adding these events.
# This is simply used to prefill the get_current_state_ids
# cache
current_state_for_room = { }
# map room_id->list[event_ids] giving the new forward
# extremities in each room
new_forward_extremeties = { }
# map room_id->(to_delete, to_insert) where to_delete is a list
# of type/state keys to remove from current state, and to_insert
# is a map (type,key)->event_id giving the state delta in each
# room
state_delta _for_room = { }
# map room_id->(type,state_key)->event_id tracking the full
# state in each room after adding these events.
# This is simply used to prefill the get_current_state_ids
# cache
current_state_for_room = { }
if not backfilled :
with Measure ( self . _clock , " _calculate_state_and_extrem " ) :
# Work out the new "current state" for each room.
# We do this by working out what the new extremities are and then
# calculating the state from that.
events_by_room = { }
for event , context in chunk :
events_by_room . setdefault ( event . room_id , [ ] ) . append (
( event , context )
)
# map room_id->(to_delete, to_insert) where to_delete is a list
# of type/state keys to remove from current state, and to_insert
# is a map (type,key)->event_id giving the state delta in each
# room
state_delta_for_room = { }
for room_id , ev_ctx_rm in iteritems ( events_by_room ) :
latest_event_ids = yield self . get_latest_event_ids_in_room (
room_id
)
new_latest_event_ids = yield self . _calculate_new_extremities (
room_id , ev_ctx_rm , latest_event_ids
if not backfilled :
with Measure ( self . _clock , " _calculate_state_and_extrem " ) :
# Work out the new "current state" for each room.
# We do this by working out what the new extremities are and then
# calculating the state from that.
events_by_room = { }
for event , context in chunk :
events_by_room . setdefault ( event . room_id , [ ] ) . append (
( event , context )
)
for room_id , ev_ctx_rm in iteritems ( events_by_room ) :
latest_event_ids = yield self . get_latest_event_ids_in_room (
room_id
)
new_latest_event_ids = yield self . _calculate_new_extremities (
room_id , ev_ctx_rm , latest_event_ids
)
latest_event_ids = set ( latest_event_ids )
if new_latest_event_ids == latest_event_ids :
# No change in extremities, so no change in state
continue
# there should always be at least one forward extremity.
# (except during the initial persistence of the send_join
# results, in which case there will be no existing
# extremities, so we'll `continue` above and skip this bit.)
assert new_latest_event_ids , " No forward extremities left! "
new_forward_extremeties [ room_id ] = new_latest_event_ids
len_1 = (
len ( latest_event_ids ) == 1
and len ( new_latest_event_ids ) == 1
)
if len_1 :
all_single_prev_not_state = all (
len ( event . prev_event_ids ( ) ) == 1
and not event . is_state ( )
for event , ctx in ev_ctx_rm
)
latest_event_ids = set ( latest_event_ids )
if new_latest_event_ids == latest_event_ids :
# No change in extremities, so no change in state
# Don't bother calculating state if they're just
# a long chain of single ancestor non-state events.
if all_single_prev_not_state :
continue
# there should always be at least one forward extremity.
# (except during the initial persistence of the send_join
# results, in which case there will be no existing
# extremities, so we'll `continue` above and skip this bit.)
assert new_latest_event_ids , " No forward extremities left! "
new_forward_extremeties [ room_id ] = new_latest_event_ids
len_1 = (
len ( latest_event_ids ) == 1
and len ( new_latest_event_ids ) == 1
state_delta_counter . inc ( )
if len ( new_latest_event_ids ) == 1 :
state_delta_single_event_counter . inc ( )
# This is a fairly handwavey check to see if we could
# have guessed what the delta would have been when
# processing one of these events.
# What we're interested in is if the latest extremities
# were the same when we created the event as they are
# now. When this server creates a new event (as opposed
# to receiving it over federation) it will use the
# forward extremities as the prev_events, so we can
# guess this by looking at the prev_events and checking
# if they match the current forward extremities.
for ev , _ in ev_ctx_rm :
prev_event_ids = set ( ev . prev_event_ids ( ) )
if latest_event_ids == prev_event_ids :
state_delta_reuse_delta_counter . inc ( )
break
logger . info ( " Calculating state delta for room %s " , room_id )
with Measure (
self . _clock , " persist_events.get_new_state_after_events "
) :
res = yield self . _get_new_state_after_events (
room_id ,
ev_ctx_rm ,
latest_event_ids ,
new_latest_event_ids ,
)
if len_1 :
all_single_prev_not_state = all (
len ( event . prev_event_ids ( ) ) == 1
and not event . is_state ( )
for event , ctx in ev_ctx_rm
)
# Don't bother calculating state if they're just
# a long chain of single ancestor non-state events.
if all_single_prev_not_state :
continue
state_delta_counter . inc ( )
if len ( new_latest_event_ids ) == 1 :
state_delta_single_event_counter . inc ( )
# This is a fairly handwavey check to see if we could
# have guessed what the delta would have been when
# processing one of these events.
# What we're interested in is if the latest extremities
# were the same when we created the event as they are
# now. When this server creates a new event (as opposed
# to receiving it over federation) it will use the
# forward extremities as the prev_events, so we can
# guess this by looking at the prev_events and checking
# if they match the current forward extremities.
for ev , _ in ev_ctx_rm :
prev_event_ids = set ( ev . prev_event_ids ( ) )
if latest_event_ids == prev_event_ids :
state_delta_reuse_delta_counter . inc ( )
break
logger . info ( " Calculating state delta for room %s " , room_id )
current_state , delta_ids = res
# If either are not None then there has been a change,
# and we need to work out the delta (or use that
# given)
if delta_ids is not None :
# If there is a delta we know that we've
# only added or replaced state, never
# removed keys entirely.
state_delta_for_room [ room_id ] = ( [ ] , delta_ids )
elif current_state is not None :
with Measure (
self . _clock , " persist_events.get_new_state_after_events "
self . _clock , " persist_events.calculate_state_delta "
) :
res = yield self . _get_new_state_after_events (
room_id ,
ev_ctx_rm ,
latest_event_ids ,
new_latest_event_ids ,
delta = yield self . _calculate_state_delta (
room_id , current_state
)
current_state , delta_ids = res
# If either are not None then there has been a change,
# and we need to work out the delta (or use that
# given)
if delta_ids is not None :
# If there is a delta we know that we've
# only added or replaced state, never
# removed keys entirely.
state_delta_for_room [ room_id ] = ( [ ] , delta_ids )
elif current_state is not None :
with Measure (
self . _clock , " persist_events.calculate_state_delta "
) :
delta = yield self . _calculate_state_delta (
room_id , current_state
)
state_delta_for_room [ room_id ] = delta
# If we have the current_state then lets prefill
# the cache with it.
if current_state is not None :
current_state_for_room [ room_id ] = current_state
state_delta_for_room [ room_id ] = delta
# If we have the current_state then lets prefill
# the cache with it.
if current_state is not None :
current_state_for_room [ room_id ] = current_state
# We want to calculate the stream orderings as late as possible, as
# we only notify after all events with a lesser stream ordering have
# been persisted. I.e. if we spend 10s inside the with block then
# that will delay all subsequent events from being notified about.
# Hence why we do it down here rather than wrapping the entire
# function.
#
# Its safe to do this after calculating the state deltas etc as we
# only need to protect the *persistence* of the events. This is to
# ensure that queries of the form "fetch events since X" don't
# return events and stream positions after events that are still in
# flight, as otherwise subsequent requests "fetch event since Y"
# will not return those events.
#
# Note: Multiple instances of this function cannot be in flight at
# the same time for the same room.
if backfilled :
stream_ordering_manager = self . _backfill_id_gen . get_next_mult (
len ( chunk )
)
else :
stream_ordering_manager = self . _stream_id_gen . get_next_mult ( len ( chunk ) )
with stream_ordering_manager as stream_orderings :
for ( event , context ) , stream in zip ( chunk , stream_orderings ) :
event . internal_metadata . stream_ordering = stream
yield self . runInteraction (
" persist_events " ,
Erik Johnston
5 years ago