import { EventEmitter } from 'events' ; import { cpus } from 'os' ; import { isMaster , setupMaster } from 'cluster' ; import { Cluster } from '../clusters/Cluster' ; import { BaseClusterWorker } from '../clusters/BaseClusterWorker' ; import { Service , ServiceOptions } from '../services/Service' ; import { BaseServiceWorker } from '../services/BaseServiceWorker' ; import { MasterIPC } from './MasterIPC' ; import { SharderEvents , IPCEvents } from '../util/constants' ; import { ClientOptions } from 'eris' ; import { MeganeStats , IPCResult } from '../' ; import fetch from 'node-fetch' ; export interface SharderOptions { /** Path to the js file for clusters to run */ path : string ; /** Discord bot token */ token : string ; /** Number of guilds each shard should have (at initial sharding) (Only used if shardCount is set to auto) */ guildsPerShard ? : number ; /** Number of shards to create */ shardCount ? : number | 'auto' | 'auto-lbs' ; /** Maximum number of clusters to create */ clusterCount ? : number ; /** Options to pass to the Eris client constructor */ clientOptions ? : ClientOptions ; /** How long to wait for a cluster to connect before throwing an error, multiplied by the number of shards */ timeout ? : number ; /** An array of arguments to pass to the cluster node processes */ nodeArgs ? : string [ ] ; /** The socket/port for IPC to run on */ ipcSocket ? : string | number ; /** How often to update stats (in milliseconds) */ statsInterval ? : number ; /** Use custom guild member caching. Not compatible with eris' `getAllUsers` */ cacheAllMembers ? : boolean ; } interface SessionObject { url : string ; shards : number ; session_start_limit : { total : number ; remaining : number ; reset_after : number ; max_concurrency : number ; } ; } interface ClusterShardInfo { /** First 0-indexed shard for this cluster */ first : number ; /** Last 0-indexed shard for this cluster */ last : number ; /** Total number of shards across all clusters */ total : number ; } export class ShardManager extends EventEmitter { /** A map of clusters by their id */ public clusters ? : Map < number , Cluster > ; /** A map of services by their name */ public services ? : Map < string , Service > ; /** A generic store of data used by all workers */ public store = new Map < string , any > ( ) ; /** Path to the js file for clusters to run */ public path : string ; public token : string ; /** The intended number of guilds that each shard should have. This is only used when shardCount is set to auto. Do not set this to a low number or you may greatly overshoot your shard count and end up with hundreds of empty shards */ public guildsPerShard : number ; /** Creates a specific number of shards instead of using auto-sharding */ public shardCount : number | 'auto' | 'auto-lbs' ; /** The number of shards that can connect at the same time */ public maxConcurrency : number ; /** The number of clusters to create. By default this is the number of CPUs */ public clusterCount : number ; /** Eris client constructor options */ public clientOptions : ClientOptions ; /** Time to wait for clusters to start */ public timeout : number ; public nodeArgs ? : string [ ] ; public ipcSocket : string | number ; public readonly ipc ? : MasterIPC ; /** Megane manager, cluster, and service stats */ public stats ? : MeganeStats ; public statsInterval : number ; public cacheAllMembers : boolean ; public constructor ( options ) { super ( ) ; this . path = options . path ! ; this . token = options . token ! ; this . guildsPerShard = options . guildsPerShard || 1500 ; this . shardCount = options . shardCount || 'auto' ; this . maxConcurrency = process . env . MAX_CONCURRENCY ? parseInt ( process . env . MAX_CONCURRENCY , 10 ) : 1 ; this . clusterCount = options . clusterCount || cpus ( ) . length ; this . clientOptions = options . clientOptions || { } ; this . timeout = options . timeout || 30e3 ; this . nodeArgs = options . nodeArgs ; this . ipcSocket = options . ipcSocket || 8191 ; this . statsInterval = options . statsInterval || 60e3 ; this . cacheAllMembers = ! ! options . cacheAllMembers ; if ( isMaster ) { this . clusters = new Map < number , Cluster > ( ) ; this . services = new Map < string , Service > ( ) ; this . ipc = new MasterIPC ( this ) ; this . stats = { clusters : { } , services : { } , manager : { memory : process . memoryUsage ( ) , cpu : process . cpuUsage ( ) } } ; setInterval ( async ( ) => { try { this . debug ( 'Updating stats' ) ; const responses = await this . ipc ! . broadcast ( { op : IPCEvents . GET_STATS } , { receptive : true } ) as IPCResult [ ] ; for ( const response of responses ) { const data = response . d as any ; if ( data . type === 'cluster' ) this . stats ! . clusters [ data . source ] = data . stats ; else this . stats ! . services [ data . source ] = data . stats ; } this . stats ! . manager = { memory : process . memoryUsage ( ) , cpu : process . cpuUsage ( ) } ; this . emit ( SharderEvents . STATS_UPDATED , this . stats ) ; } catch ( error ) { this . emit ( SharderEvents . ERROR , error ) ; } } , this . statsInterval ) ; } } /**
     * On master: Creates clusters and forks the process
     *
     * On workers: Loads the provided file implementing a worker and calls init()
     */ public async spawn ( ) { if ( isMaster ) { if ( this . shardCount === 'auto' || this . shardCount === 'auto-lbs' ) { const largeBotSharding = this . shardCount === 'auto-lbs' ; const { shards , session_start_limit : { max_concurrency } } = await this . getBotGateway ( ) ; this . debug ( `Bot gateway recommended ${ shards } shards` ) ; if ( max_concurrency && max_concurrency !== 1 ) { this . maxConcurrency = max_concurrency ; this . debug ( `Maximum concurrency of ${ this . maxConcurrency } allowed` ) ; } this . shardCount = Math . ceil ( shards * ( 1000 / this . guildsPerShard ) ) ; if ( largeBotSharding ) { this . shardCount = this . shardCount / 16 ; if ( ! Number . isInteger ( this . shardCount ) ) this . shardCount = Math . floor ( this . shardCount ) + 1 ; this . shardCount *= 16 ; } this . debug ( `Using ${ this . shardCount } shards with a target of ${ this . guildsPerShard } guilds per shard` ) ; } if ( this . shardCount < this . clusterCount ) this . clusterCount = this . shardCount ; this . debug ( `Creating ${ this . clusterCount } clusters across ${ cpus ( ) . length } CPU cores` ) ; const shardInfo = this . clusterShards ( ) ; if ( this . nodeArgs ) setupMaster ( { execArgv : this . nodeArgs } ) ; const failed = [ ] ; for ( let i = 0 ; i < this . clusterCount ; i ++ ) { this . debug ( `Creating cluster ${ i } with ${ shardInfo [ i ] . last - shardInfo [ i ] . first + 1 } shards (${ shardInfo [ i ] . first }-${ shardInfo [ i ] . last })` ) ; const cluster = new Cluster ( this , { id : i , shards : shardInfo [ i ] } ) ; this . clusters ! . set ( i , cluster ) ; try { await cluster . spawn ( ) ; } catch { this . emit ( SharderEvents . ERROR , new Error ( `Cluster ${ cluster . id } failed to start` ) ) ; failed . push ( cluster ) ; } } if ( failed . length ) await this . retryFailed ( failed ) ; this . ipc ! . broadcast ( { op : IPCEvents . READY } ) ; this . emit ( SharderEvents . ALL_CLUSTERS_READY ) ; } else { let worker ; if ( process . env . SERVICE_NAME ) { const WorkerRequire = await import ( String ( process . env . SERVICE_PATH ) ) ; const Worker = WorkerRequire . default ? WorkerRequire . default : WorkerRequire ; worker = new Worker ( this ) as BaseServiceWorker ; } else { const WorkerRequire = await import ( this . path ) ; const Worker = WorkerRequire . default ? WorkerRequire . default : WorkerRequire ; worker = new Worker ( this ) as BaseClusterWorker ; } return worker . init ( ) ; } } /** Restarts all clusters */ public async restartAll ( ) { if ( ! isMaster ) throw new Error ( 'This can only be called on the master process' ) ; this . debug ( 'Restarting all clusters' ) ; for ( const cluster of this . clusters ! . values ( ) ) await cluster . respawn ( ) ; } /** Restarts a specific cluster */ public async restart ( clusterId ) { if ( ! isMaster ) throw new Error ( 'This can only be called on the master process' ) ; const cluster = this . clusters ! . get ( clusterId ) ; if ( ! cluster ) throw new Error ( 'No cluster with that id found' ) ; this . debug ( `Restarting cluster ${ clusterId }` ) ; await cluster . respawn ( ) ; } private async retryFailed ( clusters ) { if ( ! isMaster ) throw new Error ( 'This can only be called on the master process' ) ; this . debug ( `Restarting ${ clusters . length } failed clusters` ) ; const failed = [ ] ; for ( const cluster of clusters ) try { this . debug ( 'Restarting cluster ' + cluster . id ) ; await cluster . respawn ( ) ; } catch { this . debug ( `Cluster ${ cluster . id } failed to start again` ) ; failed . push ( cluster ) ; } if ( failed . length ) return await this . retryFailed ( failed ) ; } /** Register and spawn a service */ public async registerService ( path , options ) { if ( ! isMaster ) throw new Error ( 'This can only be called on the master process' ) ; if ( ! path || ! options || ! options . name || this . services ! . has ( options . name ) ) throw new Error ( 'You must provide a path to the worker and a unique name when registering a service' ) ; const service = new Service ( this , path , options ) ; this . services ! . set ( options . name , service ) ; try { await service . spawn ( ) ; } catch { throw new Error ( 'Service worker failed to start' ) ; } } public async eval ( script ) { return await eval ( script ) ; } /** Get the bot gateway response from Discord */ public async getBotGateway ( ) { if ( ! this . token ) throw new Error ( 'No token was provided!' ) ; this . debug ( 'Getting bot gateway' ) ; const res = await fetch ( 'https://discord.com/api/v8/gateway/bot' , { method : 'GET' , headers : { Authorization : `Bot ${ this . token . replace ( /^Bot / , '' ) }` } } ) ; if ( res . ok ) return res . json ( ) ; throw new Error ( res . statusText ) ; } private clusterShards ( ) { const clusters = [ ] ; const shardsPerCluster = Math . floor ( < number > this . shardCount / this . clusterCount ) ; const leftovers = < number > this . shardCount % this . clusterCount ; let current = 0 ; for ( let i = 0 ; i < this . clusterCount ; i ++ ) { clusters . push ( { first : current , last : current + shardsPerCluster - 1 + ( leftovers > i ? 1 : 0 ) , total : < number > this . shardCount } ) ; current += shardsPerCluster + ( leftovers > i ? 1 : 0 ) ; } return clusters ; } private debug ( message ) { this . emit ( SharderEvents . DEBUG , message ) ; } } export interface ShardManager { /** Emitted when a service spawns */ on ( event , listener ) : this ; /** Emitted when a service becomes ready */ on ( event , listener ) : this ; /** Emitted when a cluster spawns */ on ( event , listener ) : this ; /** Emitted when a cluster becomes ready */ on ( event , listener ) : this ; /** Emitted when a all clusters are ready */ on ( event , listener ) : this ; /** Emitted when a shard connects (before ready) */ on ( event , listener ) : this ; /** Emitted the first time a shard becomes ready */ on ( event , listener ) : this ; /** Emitted when a shard resumes */ on ( event , listener ) : this ; /** Emitted when a shard disconnects */ on ( event , listener ) : this ; /** Emitted when the manager updates the statistics object */ on ( event , listener ) : this ; /** Emitted when a cluster has finished caching all of it's members, or when all clusters have completed it */ on ( event , listener ) : this ; /** Emits debug messages */ on ( event , listener ) : this ; /** Emitted when there is an error */ on ( event , listener ) : this ; on ( event , listener ) : this ; /** Emitted when a service spawns */ once ( event , listener ) : this ; /** Emitted when a service becomes ready */ once ( event , listener ) : this ; /** Emitted when a cluster spawns */ once ( event , listener ) : this ; /** Emitted when a cluster becomes ready */ once ( event , listener ) : this ; /** Emitted when a all clusters are ready */ once ( event , listener ) : this ; /** Emitted when a shard connects (before ready) */ once ( event , listener ) : this ; /** Emitted the first time a shard becomes ready */ once ( event , listener ) : this ; /** Emitted when a shard resumes */ once ( event , listener ) : this ; /** Emitted when a shard disconnects */ once ( event , listener ) : this ; /** Emitted when the manager updates the statistics object */ once ( event , listener ) : this ; /** Emitted when a cluster has finished caching all of it's members, or when all clusters have completed it */ once ( event , listener ) : this ; /** Emits debug messages */ once ( event , listener ) : this ; /** Emitted when there is an error */ once ( event , listener ) : this ; once ( event , listener ) : this ; /** Emitted when a service spawns */ off ( event , listener ) : this ; /** Emitted when a service becomes ready */ off ( event , listener ) : this ; /** Emitted when a cluster spawns */ off ( event , listener ) : this ; /** Emitted when a cluster becomes ready */ off ( event , listener ) : this ; /** Emitted when a all clusters are ready */ off ( event , listener ) : this ; /** Emitted when a shard connects (before ready) */ off ( event , listener ) : this ; /** Emitted the first time a shard becomes ready */ off ( event , listener ) : this ; /** Emitted when a shard resumes */ off ( event , listener ) : this ; /** Emitted when a shard disconnects */ off ( event , listener ) : this ; /** Emitted when the manager updates the statistics object */ off ( event , listener ) : this ; /** Emitted when a cluster has finished caching all of it's members, or when all clusters have completed it */ off ( event , listener ) : this ; /** Emits debug messages */ off ( event , listener ) : this ; /** Emitted when there is an error */ off ( event , listener ) : this ; off ( event , listener ) : this ; /** Emitted when a service spawns */ emit ( event , service ) : boolean ; /** Emitted when a service becomes ready */ emit ( event , service ) : boolean ; /** Emitted when a cluster spawns */ emit ( event , cluster ) : boolean ; /** Emitted when a cluster becomes ready */ emit ( event , cluster ) : boolean ; /** Emitted when a all clusters are ready */ emit ( event ) : boolean ; /** Emitted when a shard connects (before ready) */ emit ( event , clusterId , shardId ) : boolean ; /** Emitted the first time a shard becomes ready */ emit ( event , clusterId , shardId ) : boolean ; /** Emitted when a shard resumes */ emit ( event , clusterId , shardId ) : boolean ; /** Emitted when a shard disconnects */ emit ( event , clusterId , shardId , error ) : boolean ; /** Emitted when the manager updates the statistics object */ emit ( event , stats ) : boolean ; /** Emitted when a cluster has finished caching all of it's members, or when all clusters have completed it */ emit ( event , shardId ? ) : boolean ; /** Emits debug messages */ emit ( event , message ) : boolean ; /** Emitted when there is an error */ emit ( event , error , clusterId , shardId ) : boolean ; emit ( event , ... args ) : boolean ; }