添加自定义UdpService

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<Node>();
             TokenManager = new EasyTokenManager();
-            Queue = new Queue<KeyValuePair<InfoHash, IPEndPoint>>();
         }
         private object locker = new object();
         public IMetaDataFilter Filter { get; set; }
@@ -41,7 +42,6 @@ namespace Spider.Core
 
         public HashSet<Node> KTable { get; set; }
 
-        public Queue<KeyValuePair<InfoHash, IPEndPoint>> 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, IPEndPoint>(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)
         {

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<Int32> freeIndexPool;
+
+        public BufferManager(Int32 numsize, Int32 buffersize)
+        {
+            this.numSize = numsize;
+            this.bufferSize = buffersize;
+
+        }
+
+        public void Inint()
+        {
+            buffer = new byte[numSize];
+            freeIndexPool = new Stack<int>(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;
+        }
+    }
+}

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
+{
+    /// <summary>
+    /// Based on example from http://msdn2.microsoft.com/en-us/library/system.net.sockets.socketasynceventargs.socketasynceventargs.aspx
+    /// Represents a collection of reusable SocketAsyncEventArgs objects.  
+    /// </summary>
+    internal sealed class SocketAsyncEventArgsPool
+    {
+        /// <summary>
+        /// SocketAsyncEventArgs栈
+        /// </summary>
+        Stack<SocketAsyncEventArgs> pool;
+
+        /// <summary>
+        /// 初始化SocketAsyncEventArgs池
+        /// </summary>
+        /// <param name="capacity">最大可能使用的SocketAsyncEventArgs对象.</param>
+        internal SocketAsyncEventArgsPool(Int32 capacity)
+        {
+            this.pool = new Stack<SocketAsyncEventArgs>(capacity);
+        }
+
+        /// <summary>
+        /// 返回SocketAsyncEventArgs池中的 数量
+        /// </summary>
+        internal Int32 Count
+        {
+            get { return this.pool.Count; }
+        }
+
+        /// <summary>
+        /// 弹出一个SocketAsyncEventArgs
+        /// </summary>
+        /// <returns>SocketAsyncEventArgs removed from the pool.</returns>
+        internal SocketAsyncEventArgs Pop()
+        {
+            lock (this.pool)
+            {
+                return this.pool.Pop();
+            }
+        }
+
+        /// <summary>
+        /// 添加一个 SocketAsyncEventArgs
+        /// </summary>
+        /// <param name="item">SocketAsyncEventArgs instance to add to the pool.</param>
+        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);
+            }
+        }
+    }
+}

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; } }
+
+        /// <summary>
+        /// 数据包管理
+        /// </summary>
+        private BufferManager BuffManagers;
+
+        /// <summary>
+        /// Socket异步对象池
+        /// </summary>
+        private SocketAsyncEventArgsPool SocketAsynPool;
+        /// <summary>
+        /// 接收包大小
+        /// </summary>
+        private int MaxBufferSize;
+        /// <summary>
+        /// 最大用户连接
+        /// </summary>
+        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<SocketAsyncEventArgs>(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 资源释放
+        /// <summary>
+        /// 用来确定是否以释放
+        /// </summary>
+        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
+
+    }
+}

Spider/Spider.csproj

@@ -48,6 +48,9 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="Core\UdpServer\BufferManager.cs" />
+    <Compile Include="Core\UdpServer\SocketAsyncEventArgsPool.cs" />
+    <Compile Include="Core\UdpServer\UDPService.cs" />
     <Compile Include="Queue\IQueue.cs" />
     <Compile Include="Core\DHTSpider.cs" />
     <Compile Include="Core\EasyTokenManager.cs" />
@@ -68,5 +71,6 @@
   <ItemGroup>
     <None Include="packages.config" />
   </ItemGroup>
+  <ItemGroup />
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
 </Project>

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);
+                }
             }
         }
     }

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}");
             }
         }