j)dZddlmZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z mZddlmZeegefZGddZGdd ZGd d ZdS) z Distributed Chat Cluster Router with Sharded Ring Buffers for 1,000,000 CCU. Supports Hybrid Redis Cluster Pub/Sub bridging and In-Memory high-throughput sharding. ) annotationsN)AnyCallable CoroutineDictListOptionalSetTuple)ChatMessageDTOc\eZdZdZdddZdd Zdd ZddZddZd dZ d!dZ d"dZ d#dZ dS)$ShardedChannelRegistryz Sharded subscriber registry that minimizes lock contention when managing up to 1,000,000 concurrent client subscriptions. @ num_shardsintreturnNonec||_dt|D|_dt|D|_dt|D|_dS)Ncg|]}iSr.0_s 8C:\Projects\FreeExile\server\chat\chat_cluster_router.py z3ShardedChannelRegistry.__init__..FeFeFearFeFeFecg|]}iSrrrs rrz3ShardedChannelRegistry.__init__..rrcg|]}iSrrrs rrz3ShardedChannelRegistry.__init__..sGfGfGfqGfGfGfr)rrangeshards cached_sync cached_async)selfrs r__init__zShardedChannelRegistry.__init__sn$FeFeSXYcSdSdFeFeFe FeFeSXYcSdSdFeFeFeGfGfTYZdTeTeGfGfGfr channel_keystrc0t||jzSN)hashrr$r&s r_get_shard_indexz'ShardedChannelRegistry._get_shard_index!sK  4?22r shard_idxcg}g}|j||D]A}tj|r||,||B||j||<||j||<dSr))r!valuesinspectiscoroutinefunctionappendr"r#)r$r-r& sync_list async_listcbs r_rebuild_cachez%ShardedChannelRegistry._rebuild_cache$s.0 /1 +i(5<<>> % %B*2.. %!!"%%%%  $$$$3<#K04>)$[111r client_idcallbackSubscriberCallbackc||}|j|}||vri||<||||<|||dSr))r,r!r6)r$r&r7r8r-shards r subscribez ShardedChannelRegistry.subscribe/sa))+66  I& e # #!#E+ (0k9% I{33333rcX||}|j|}||vrdS|||d||sG||=|j||d|j||ddS|||dSr))r,r!popr"r#r6)r$r&r7r-r;s r unsubscribez"ShardedChannelRegistry.unsubscribe7s))+66  I& e # # F ky$///[! 8k"  Y ' + +K > > >  i ( , ,[$ ? ? ? ? ?    ; 7 7 7 7 7r9Tuple[List[SubscriberCallback], List[SubscriberCallback]]c||}|j||g|j||gfSr)r,r"getr#)r$r&r-s rget_partitioned_callbacksz0ShardedChannelRegistry.get_partitioned_callbacksDsU))+66  Y ' + +K < <  i ( , ,[" = =  rList[SubscriberCallback]c||}|j||g}|j||g}||zSr)rB)r$r&r-sync_cbs async_cbss r get_callbacksz$ShardedChannelRegistry.get_callbacksKsZ))+66 #I.22;CC%i044["EE )##r$List[Tuple[int, SubscriberCallback]]c||}|j|}||}|r!t|ngSr))r,r!rClistitems)r$r&r-r; subscriberss rget_subscribersz&ShardedChannelRegistry.get_subscribersQsV))+66  I&ii ,, ,7?tK%%''(((R?rcpd}|jD]+}|D]}|t|z },|S)Nr)r!r/len)r$countr;subss rtotal_subscribers_countz.ShardedChannelRegistry.total_subscribers_countWsK[ # #E  # #T" # rN)r)rrrr)r&r'rr)r-rr&r'rr)r&r'r7rr8r9rr)r&r'r7rrr)r&r'rr@)r&r'rrE)r&r'rrJ)rr) __name__ __module__ __qualname____doc__r%r,r6r<r?rDrIrOrTrrrrrs ggggg3333 ? ? ? ?4444 8 8 8 8    $$$$ @@@@ rrcPeZdZdZ ddd Zedd ZddZddZddZ dS)RedisShardedPubSubBridgez Redis 7 Sharded Pub/Sub bridge interface (SPUBLISH / SSUBSCRIBE). Routes messages using Redis cluster hash tags {...} to confine pub/sub traffic to the slot-owning node, preventing cluster-wide broadcast storms. Npublish_handler8Optional[Callable[[str, str], Coroutine[Any, Any, int]]]subscribe_handlerHOptional[Callable[[str, SubscriberCallback], Coroutine[Any, Any, None]]]rrc0||_||_g|_dSr))_publish_handler_subscribe_handlerpublished_messages)r$r[r]s rr%z!RedisShardedPubSubBridge.__init__fs" !0"39;rr&r'cf|dr|dr|Sd|dS)z Wraps channel key into Redis Cluster hash tag format. e.g. 'world' -> '{world}', 'zone:barrow' -> '{zone:barrow}', 'guild:101' -> '{guild:101}'. {}) startswithendswith)r&s rto_sharded_channelz+RedisShardedPubSubBridge.to_sharded_channelosF  ! !# & & ;+?+?+D+D  #K####rmessageChatMessageDTO | strrcK||}t|tr|n|j}|j||f|j|||d{VSdS)z2Dispatches SPUBLISH command to Redis 7 shard slot.N)rh isinstancer' raw_contentrbr2r`)r$r&ri sharded_chpayloads rspublishz!RedisShardedPubSubBridge.spublishys,,[99 '55N''7;N && G'<===  ,..z7CCCCCCCC Cqrr8r9c~K||}|j|||d{VdSdS)z2Registers SSUBSCRIBE on the Redis 7 shard channel.N)rhra)r$r&r8ros r ssubscribez#RedisShardedPubSubBridge.ssubscribesX,,[99  " .))*h?? ? ? ? ? ? ? ? ? ? / .rc KdS)z Removes SSUBSCRIBE registration.Nrr+s r sunsubscribez%RedisShardedPubSubBridge.sunsubscribes  r)NN)r[r\r]r^rr)r&r'rr')r&r'rirjrr)r&r'r8r9rr)r&r'rr) rUrVrWrXr% staticmethodrhrqrsrurrrrZrZ_sUYfj<<<<<$$$\$@@@@       rrZcJeZdZdZ ddd Zd dZd!dZd"dZd#dZd$dZ dS)%ChatClusterRouterz Core distributed fan-out engine capable of horizontal cluster scaling. Combines local sharded non-blocking delivery with Redis Pub/Sub cluster bridge. rNrrbatch_fanout_size redis_bridge"Optional[RedisShardedPubSubBridge]rrcht||_||_||_g|_d|_dS)N)rr)rregistryrzr{_dispatch_latencies_ms_total_messages_routed)r$rrzr{s rr%zChatClusterRouter.__init__s< /*EEE !2(35#+,###rbridgerZc||_dS)z4Attaches an external Redis 7 Sharded Pub/Sub bridge.N)r{)r$rs rattach_redis_bridgez%ChatClusterRouter.attach_redis_bridges"rr5r9rir /Tuple[bool, Optional[Coroutine[Any, Any, Any]]]c tj|r||}d|fS||}tj|rd|fSdS#t$rYdSwxYw)a1 Invokes subscriber callback safely: If cb is a coroutine function (async def), returns coroutine for batch gathering. If cb is a regular synchronous callable, invokes it directly without coroutine overhead. Eliminates unneeded coroutines and TypeError exceptions. T)TN)FN)r0r1asyncio iscoroutine Exception)r$r5ricororess r _safe_invokezChatClusterRouter._safe_invokes *2.. "r'{{Tz!"W++C"3'' !Sy :   ;; s"A "A AAr&r'cNKtj}|j|\}}d}|D]O} ||}|*t j|r||n|dz }@#t$rYLwxYw|rg} |D]j} ||} | d| | n #t$r|dz }Ynt$rYnwxYw[#t$rYgwxYw| r3t j | ddid{V} |td| Dz }|j 3 |j ||d{Vn#t$rYnwxYwtj|z dz} || |xjdz c_|S)z Dispatches a chat message to all subscribers of a channel key. Executes fan-out in parallel batches to achieve ultra-low p99 latency. rNrlreturn_exceptionsTc3DK|]}t|tdVdS)rlN)rmr)rrs r z9ChatClusterRouter.broadcast_to_channel..s1&Z&ZQAyAYAY&Zq&Z&Z&Z&Z&Z&Zrg@@)time perf_counterr~rDrrr2rsend StopIterationgathersumr{rq_record_latencyr) r$r&rit_startrGrHdelivered_countr5rpending_coroutinesrresults elapsed_mss rbroadcast_to_channelz&ChatClusterRouter.broadcast_to_channelsl#%%"mEEkRR)  B bkk?w':3'?'??$$R((((#q(O      [AC    2g;;D 8 $+11$7777 )---'1,$ !D" [ '0B [VZ [ [[[[[[[3&Z&Z'&Z&Z&Z#Z#ZZ   ( '00gFFFFFFFFFF    '))G3v=  Z((( ##q(##ser2)r$rs rrz!ChatClusterRouter._record_latencysL t* + +t 3 3  ' + +A . . . #**:66666rDict[str, float]c|jsdddddSt|j}t|}|t|dz}|t t|dz|dz }|t t|dz|dz }t ||z }t |dt |dt |dt |d|jdS) z;Calculates p50, p95, p99 latency metrics for observability.g)p50p95p99avgg?gffffff?rlgGz?)rrrr total_routed)rsortedrQrminrroundr)r$sorted_latenciesnrrrrs rget_latency_statsz#ChatClusterRouter.get_latency_statss* Ds3sCC C!$"=>>  ! !s1t8}}-s3q4x==!a%889s3q4x==!a%889"##a'a==a==a==a== 7    r)rryN)rrrzrr{r|rr)rrZrr)r5r9rir rr)r&r'rir rr)rrrr)rr) rUrVrWrXr%rrrrrrrrrxrxs!%;? - - - - -####,9999v7777       rrx)rX __future__rrr0rtypingrrrrrr r r server.chat.chat_typesr r9rrZrxrrrrs7 #""""" MMMMMMMMMMMMMMMMMMMM111111~.34IIIIIIIIX+ + + + + + + + \~ ~ ~ ~ ~ ~ ~ ~ ~ ~ r