diff --git a/Spider/Core/DHTSpider.cs b/Spider/Core/DHTSpider.cs index 28fcf08..47144cb 100644 --- a/Spider/Core/DHTSpider.cs +++ b/Spider/Core/DHTSpider.cs @@ -1,4 +1,5 @@ -using Spider.Log; +using Spider.Core.UdpServer; +using Spider.Log; using System; using System.Collections.Generic; using System.IO; @@ -27,10 +28,10 @@ namespace Spider.Core { LocalId = NodeId.Create(); - listener = new DhtListener(localAddress); + //listener = new DhtListener(localAddress); + udp = new UDPService(localAddress); KTable = new HashSet(); TokenManager = new EasyTokenManager(); - Queue = new Queue>(); } private object locker = new object(); public IMetaDataFilter Filter { get; set; } @@ -41,7 +42,6 @@ namespace Spider.Core public HashSet KTable { get; set; } - public Queue> Queue { get; set; } private bool disposed = false; public bool Disposed { @@ -68,13 +68,14 @@ namespace Spider.Core public void Add(Node node) { + Logger.Fatal($"Add1 {KTable.Count} {node.Id} {node.Token} {node.EndPoint}"); lock (locker) { if (!KTable.Contains(node)) { lock (locker) { - Logger.Fatal($"Add {KTable.Count} {node.Id} {node.Token} {node.EndPoint}"); + Logger.Fatal($"Add2 {KTable.Count} {node.Id} {node.Token} {node.EndPoint}"); KTable.Add(node); } } @@ -94,7 +95,6 @@ namespace Spider.Core if (Filter == null || (Filter != null && Filter.Ignore(infohash))) { NewMetadata?.Invoke(this, new NewMetadataEventArgs(infohash, endpoint)); - Queue.Enqueue(new KeyValuePair(infohash, endpoint)); } } catch (Exception) @@ -143,13 +143,16 @@ namespace Spider.Core public void Send(DhtMessage msg, IPEndPoint endpoint) { var buffer = msg.Encode(); - listener.Send(buffer, endpoint); + //listener.Send(buffer, endpoint); + udp.Send(endpoint, buffer); } public void Start() { - listener.Start(); - listener.MessageReceived += OnMessageReceived; + //listener.Start(); + //listener.MessageReceived += OnMessageReceived; + udp.Start(); + udp.MessageReceived += OnMessageReceived; Task.Run(() => { @@ -168,7 +171,7 @@ namespace Spider.Core } public void Stop() { - listener.Stop(); + //listener.Stop(); } private void JoinDHTNetwork() @@ -211,7 +214,7 @@ namespace Spider.Core } private DhtListener listener; - + private UDPService udp; private void OnMessageReceived(byte[] buffer, IPEndPoint endpoint) { diff --git a/Spider/Core/UdpServer/BufferManager.cs b/Spider/Core/UdpServer/BufferManager.cs new file mode 100644 index 0000000..4094fe0 --- /dev/null +++ b/Spider/Core/UdpServer/BufferManager.cs @@ -0,0 +1,56 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net.Sockets; +using System.Text; +using System.Threading.Tasks; + +namespace Spider.Core.UdpServer +{ + internal sealed class BufferManager + { + private Byte[] buffer; + private Int32 bufferSize; + private Int32 numSize; + private Int32 currentIndex; + private Stack freeIndexPool; + + public BufferManager(Int32 numsize, Int32 buffersize) + { + this.numSize = numsize; + this.bufferSize = buffersize; + + } + + public void Inint() + { + buffer = new byte[numSize]; + freeIndexPool = new Stack(numSize / bufferSize); + } + + internal void FreeBuffer(SocketAsyncEventArgs args) + { + freeIndexPool.Push(args.Offset); + args.SetBuffer(null, 0, 0); + + } + + internal Boolean SetBuffer(SocketAsyncEventArgs args) + { + if (this.freeIndexPool.Count > 0) + { + args.SetBuffer(this.buffer, this.freeIndexPool.Pop(), this.bufferSize); + } + else + { + if ((this.numSize - this.bufferSize) < this.currentIndex) + { + return false; + } + args.SetBuffer(this.buffer, this.currentIndex, this.bufferSize); + this.currentIndex += this.bufferSize; + } + return true; + } + } +} diff --git a/Spider/Core/UdpServer/SocketAsyncEventArgsPool.cs b/Spider/Core/UdpServer/SocketAsyncEventArgsPool.cs new file mode 100644 index 0000000..584fcc3 --- /dev/null +++ b/Spider/Core/UdpServer/SocketAsyncEventArgsPool.cs @@ -0,0 +1,66 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net.Sockets; +using System.Text; +using System.Threading.Tasks; + +namespace Spider.Core.UdpServer +{ + /// + /// Based on example from http://msdn2.microsoft.com/en-us/library/system.net.sockets.socketasynceventargs.socketasynceventargs.aspx + /// Represents a collection of reusable SocketAsyncEventArgs objects. + /// + internal sealed class SocketAsyncEventArgsPool + { + /// + /// SocketAsyncEventArgs栈 + /// + Stack pool; + + /// + /// 初始化SocketAsyncEventArgs池 + /// + /// 最大可能使用的SocketAsyncEventArgs对象. + internal SocketAsyncEventArgsPool(Int32 capacity) + { + this.pool = new Stack(capacity); + } + + /// + /// 返回SocketAsyncEventArgs池中的 数量 + /// + internal Int32 Count + { + get { return this.pool.Count; } + } + + /// + /// 弹出一个SocketAsyncEventArgs + /// + /// SocketAsyncEventArgs removed from the pool. + internal SocketAsyncEventArgs Pop() + { + lock (this.pool) + { + return this.pool.Pop(); + } + } + + /// + /// 添加一个 SocketAsyncEventArgs + /// + /// SocketAsyncEventArgs instance to add to the pool. + internal void Push(SocketAsyncEventArgs item) + { + if (item == null) + { + throw new ArgumentNullException("Items added to a SocketAsyncEventArgsPool cannot be null"); + } + lock (this.pool) + { + this.pool.Push(item); + } + } + } +} diff --git a/Spider/Core/UdpServer/UDPService.cs b/Spider/Core/UdpServer/UDPService.cs new file mode 100644 index 0000000..57c1ecc --- /dev/null +++ b/Spider/Core/UdpServer/UDPService.cs @@ -0,0 +1,231 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace Spider.Core.UdpServer +{ + public delegate void MessageReceived(byte[] buffer, IPEndPoint endpoint); + + public class UDPService : IDisposable + { + private IPEndPoint endpoint; + public IPEndPoint Endpoint + { + get { return endpoint; } + } + + private Socket sock; + public Socket Sock { get { return sock; } } + + /// + /// 数据包管理 + /// + private BufferManager BuffManagers; + + /// + /// Socket异步对象池 + /// + private SocketAsyncEventArgsPool SocketAsynPool; + /// + /// 接收包大小 + /// + private int MaxBufferSize; + /// + /// 最大用户连接 + /// + private int MaxConnectCout; + + private AutoResetEvent[] reset; + + public Thread ReceiveThread { get; private set; } + + public UDPService(IPEndPoint endpoint) + { + this.endpoint = endpoint; + + this.MaxBufferSize = 1024;//数据包最大缓冲区 + this.MaxConnectCout = 10000;//最大连接数 + + this.reset = new AutoResetEvent[1]; + reset[0] = new AutoResetEvent(false); + } + + public void Start() + { + Run(); + } + + private void Run() + { + if (isDisposed == true) + { + throw new ObjectDisposedException("UDPService is Disposed"); + } + + sock = new Socket(endpoint.AddressFamily, SocketType.Dgram, ProtocolType.Udp); + sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, 1); + sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Broadcast, true); + sock.DontFragment = true; + sock.EnableBroadcast = true; + sock.Bind(endpoint); + + BuffManagers = new BufferManager(MaxConnectCout * MaxBufferSize, MaxBufferSize); + BuffManagers.Inint(); + + SocketAsynPool = new SocketAsyncEventArgsPool(MaxConnectCout); + + for (int i = 0; i < MaxConnectCout; i++) + { + SocketAsyncEventArgs socketasyn = new SocketAsyncEventArgs(); + socketasyn.RemoteEndPoint = new IPEndPoint(IPAddress.Any, 0); + socketasyn.Completed += new EventHandler(Asyn_Completed); + SocketAsynPool.Push(socketasyn); + } + + reset[0].Set(); + Receive(); + } + + private void Receive() + { + ReceiveThread = new System.Threading.Thread(new System.Threading.ThreadStart(() => + { + while (true) + { + System.Threading.WaitHandle.WaitAll(reset); + reset[0].Set(); + + if (SocketAsynPool.Count > 0) + { + SocketAsyncEventArgs sockasyn = SocketAsynPool.Pop(); + + if (BuffManagers.SetBuffer(sockasyn)) + { + if (!Sock.ReceiveFromAsync(sockasyn)) + { + BeginReceive(sockasyn); + } + + } + } + else + { + reset[0].Reset(); + } + + } + })); + + ReceiveThread.Start(); + } + + private void Asyn_Completed(object sender, SocketAsyncEventArgs e) + { + switch (e.LastOperation) + { + case SocketAsyncOperation.ReceiveFrom: + case SocketAsyncOperation.Receive: + BeginReceive(e); + break; + + } + + e.AcceptSocket = null; + BuffManagers.FreeBuffer(e); + SocketAsynPool.Push(e); + reset[0].Set(); + } + + public event MessageReceived MessageReceived; + + private void BeginReceive(SocketAsyncEventArgs e) + { + if (e.SocketError == SocketError.Success && e.BytesTransferred > 0) + { + byte[] data = new byte[e.BytesTransferred]; + + Buffer.BlockCopy(e.Buffer, e.Offset, data, 0, data.Length); + + MessageReceived?.Invoke(data, (IPEndPoint)e.RemoteEndPoint); + + + } + } + + public void Send(IPEndPoint ipendpoint, byte[] data) + { + sock.BeginSendTo(data, 0, data.Length, SocketFlags.None, ipendpoint, SendCallBack, sock); + } + + private void SendCallBack(IAsyncResult result) + { + try + { + Socket sock = result.AsyncState as Socket; + + if (sock != null) + { + sock.EndSend(result); + } + } + catch + { + + } + } + + #region 资源释放 + /// + /// 用来确定是否以释放 + /// + private bool isDisposed; + + + ~UDPService() + { + this.Dispose(false); + + } + + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + protected virtual void Dispose(bool disposing) + { + if (!isDisposed || disposing) + { + try + { + sock.Shutdown(SocketShutdown.Both); + sock.Close(); + + for (int i = 0; i < SocketAsynPool.Count; i++) + { + SocketAsyncEventArgs args = SocketAsynPool.Pop(); + + BuffManagers.FreeBuffer(args); + } + + + ReceiveThread.Abort(); + + } + catch + { + } + + isDisposed = true; + } + } + #endregion + + } +} diff --git a/Spider/Spider.csproj b/Spider/Spider.csproj index dd3d66b..1420022 100644 --- a/Spider/Spider.csproj +++ b/Spider/Spider.csproj @@ -48,6 +48,9 @@ + + + @@ -68,5 +71,6 @@ + \ No newline at end of file diff --git a/Spider/SpiderConfiguration.cs b/Spider/SpiderConfiguration.cs index 354b35f..c9fc19d 100644 --- a/Spider/SpiderConfiguration.cs +++ b/Spider/SpiderConfiguration.cs @@ -114,7 +114,7 @@ namespace Spider spider.NewMetadata += DHTSpider_NewMetadata; spider.Start(); }); - Thread.Sleep(1000 * 2); + Thread.Sleep(1000); } for (var i = 0; i < _option.MaxDownLoadThreadCount; i++) @@ -125,7 +125,7 @@ namespace Spider { Download(id); }); - Thread.Sleep(1000); + Thread.Sleep(500); } @@ -183,6 +183,10 @@ namespace Spider { Logger.Error($"Download {ex.Message}"); } + finally + { + Thread.Sleep(100); + } } } } diff --git a/Tancoder.Torrent/Listeners/UdpListener.cs b/Tancoder.Torrent/Listeners/UdpListener.cs index b57f919..5872ae5 100644 --- a/Tancoder.Torrent/Listeners/UdpListener.cs +++ b/Tancoder.Torrent/Listeners/UdpListener.cs @@ -61,9 +61,10 @@ namespace Tancoder.Torrent OnMessageReceived(buffer, e); client.BeginReceive(EndReceive, null); } - catch (ObjectDisposedException) + catch (ObjectDisposedException ex) { // Ignore, we're finished! + throw new Exception($"UdpListener ObjectDisposedException: {ex}"); } catch (SocketException ex) { @@ -79,17 +80,26 @@ namespace Tancoder.Torrent client.BeginReceive(EndReceive, null); return; } - catch (ObjectDisposedException) + catch (ObjectDisposedException oe) { - return; + throw new Exception($"UdpListener ObjectDisposedException: {oe}"); } catch (SocketException e) { + throw new Exception($"UdpListener SocketException: {e}"); if (e.ErrorCode != 10054) return; } } } + else if (ex.ErrorCode == 10052) + { + + } + else + { + throw new Exception($"UdpListener SocketException: {ex}"); + } } } @@ -104,7 +114,7 @@ namespace Tancoder.Torrent } catch (Exception ex) { - Debug.WriteLine("UdpListener could not send message: {0}", ex); + throw new Exception($"UdpListener could not send message: {ex}"); } }