mattburchett/NadekoBot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Shard coordinator is not indepentent from the shard 0

Browse Source
This commit is contained in:
Master Kwoth
2017-10-10 18:24:36 +02:00
parent 0bacb1f780
commit db6fa9af1a
14 changed files with 225 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
NadekoBot.Core/Services/ShardsCoordinator.cs
View File

@ -2,75 +2,130 @@
using NLog;
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using NadekoBot.Common.ShardCom;
using StackExchange.Redis;
using Newtonsoft.Json;
using NadekoBot.Extensions;
namespace NadekoBot.Services
{
public class ShardsCoordinator
{
private readonly BotCredentials _creds;
private readonly string _key;
private readonly Process[] _shardProcesses;
public ShardComMessage[] Statuses { get; }
public int GuildCount => Statuses.ToArray()
.Where(x => x != null)
.Sum(x => x.Guilds);
private readonly Logger _log;
private readonly ShardComServer _comServer;
private readonly int _curProcessId;
private readonly ConnectionMultiplexer _redis;
private ShardComMessage _defaultShardState;
public ShardsCoordinator(IDataCache cache)
public ShardsCoordinator()
{
//load main stuff
LogSetup.SetupLogger();
_log = LogManager.GetCurrentClassLogger();
_creds = new BotCredentials();
_shardProcesses = new Process[_creds.TotalShards];
Statuses = new ShardComMessage[_creds.TotalShards];
_key = _creds.RedisKey();
for (int i = 0; i < Statuses.Length; i++)
//setup initial shard statuses
_defaultShardState = new ShardComMessage()
{
Statuses[i] = new ShardComMessage();
var s = Statuses[i];
s.ConnectionState = Discord.ConnectionState.Disconnected;
s.Guilds = 0;
s.ShardId = i;
s.Time = DateTime.Now - TimeSpan.FromMinutes(1);
ConnectionState = Discord.ConnectionState.Disconnected,
Guilds = 0,
Time = DateTime.Now - TimeSpan.FromMinutes(1)
};
var db = _redis.GetDatabase();
_shardProcesses = new Process[_creds.TotalShards];
for (int i = 0; i < _creds.TotalShards; i++)
{
_defaultShardState.ShardId = i;
db.ListSetByIndex(_key + "shardstats",
i,
JsonConvert.SerializeObject(_defaultShardState),
CommandFlags.FireAndForget);
}
_log = LogManager.GetCurrentClassLogger();
_comServer = new ShardComServer(cache);
_comServer.Start();
_comServer.OnDataReceived += _comServer_OnDataReceived;
_curProcessId = Process.GetCurrentProcess().Id;
_redis = ConnectionMultiplexer.Connect("127.0.0.1");
var sub = _redis.GetSubscriber();
sub.Subscribe(_key + "_shardcoord_send",
OnDataReceived,
CommandFlags.FireAndForget);
sub.Subscribe(_key + "_shardcoord_restart",
OnRestart,
CommandFlags.FireAndForget);
sub.Subscribe(_key + "_shardcoord_stop",
OnStop,
CommandFlags.FireAndForget);
}
private Task _comServer_OnDataReceived(ShardComMessage msg)
private void OnStop(RedisChannel ch, RedisValue data)
{
Statuses[msg.ShardId] = msg;
if (msg.ConnectionState == Discord.ConnectionState.Disconnected || msg.ConnectionState == Discord.ConnectionState.Disconnecting)
_log.Error("!!! SHARD {0} IS IN {1} STATE", msg.ShardId, msg.ConnectionState.ToString());
return Task.CompletedTask;
var shardId = JsonConvert.DeserializeObject<int>(data);
var db = _redis.GetDatabase();
_defaultShardState.ShardId = shardId;
db.ListSetByIndex(_key + "shardstats",
shardId,
JsonConvert.SerializeObject(_defaultShardState),
CommandFlags.FireAndForget);
var p = _shardProcesses[shardId];
_shardProcesses[shardId] = null;
try { p?.Kill(); } catch { }
try { p?.Dispose(); } catch { }
}
private void OnRestart(RedisChannel ch, RedisValue data)
{
OnStop(ch, data);
var shardId = JsonConvert.DeserializeObject<int>(data);
_shardProcesses[shardId] = StartShard(shardId);
}
private void OnDataReceived(RedisChannel ch, RedisValue data)
{
var msg = JsonConvert.DeserializeObject<ShardComMessage>(data);
if (msg == null)
return;
var db = _redis.GetDatabase();
db.ListSetByIndex(_key + "shardstats",
msg.ShardId,
data,
CommandFlags.FireAndForget);
if (msg.ConnectionState == Discord.ConnectionState.Disconnected
|| msg.ConnectionState == Discord.ConnectionState.Disconnecting)
{
_log.Error("!!! SHARD {0} IS IN {1} STATE !!!", msg.ShardId, msg.ConnectionState.ToString());
}
return;
}
public async Task RunAsync()
{
for (int i = 1; i < _creds.TotalShards; i++)
for (int i = 0; i < _creds.TotalShards; i++)
{
var p = Process.Start(new ProcessStartInfo()
{
FileName = _creds.ShardRunCommand,
Arguments = string.Format(_creds.ShardRunArguments, i, _curProcessId, "")
});
// last "" in format is for backwards compatibility
// because current startup commands have {2} in them probably
await Task.Delay(5000);
var p = StartShard(i);
_shardProcesses[i] = p;
await Task.Delay(6000);
}
}
private Process StartShard(int shardId)
{
return Process.Start(new ProcessStartInfo()
{
FileName = _creds.ShardRunCommand,
Arguments = string.Format(_creds.ShardRunArguments, shardId, _curProcessId, "")
});
// last "" in format is for backwards compatibility
// because current startup commands have {2} in them probably
}
public async Task RunAndBlockAsync()
{
try
@ -80,14 +135,15 @@ namespace NadekoBot.Services
catch (Exception ex)
{
_log.Error(ex);
foreach (var p in _shardProcesses)
{
try { p.Kill(); } catch { }
try { p.Dispose(); } catch { }
}
return;
}
await Task.Delay(-1);
foreach (var p in _shardProcesses)
{
try { p.Kill(); } catch { }
try { p.Dispose(); } catch { }
}
}
}
}