Author: alanmc
Date: 2008-01-03 11:58:11 -0500 (Thu, 03 Jan 2008)
New Revision: 92190
Modified:
trunk/bitsharp/src/MonoTorrent/Client/BufferedFileRead.cs
trunk/bitsharp/src/MonoTorrent/Client/BufferedFileWrite.cs
trunk/bitsharp/src/MonoTorrent/Client/Managers/DiskManager.cs
trunk/bitsharp/src/MonoTorrent/Client/Managers/FileManager.cs
trunk/bitsharp/src/MonoTorrent/Client/Managers/TorrentManager.cs
Log:
Implemented initial read rate limiting. Now both read and write limiting are
supported. Huzzah!
Modified: trunk/bitsharp/src/MonoTorrent/Client/BufferedFileRead.cs
===================================================================
--- trunk/bitsharp/src/MonoTorrent/Client/BufferedFileRead.cs 2008-01-03
16:46:00 UTC (rev 92189)
+++ trunk/bitsharp/src/MonoTorrent/Client/BufferedFileRead.cs 2008-01-03
16:58:11 UTC (rev 92190)
@@ -1,10 +1,28 @@
using System;
using System.Collections.Generic;
using System.Text;
+using System.Threading;
namespace MonoTorrent.Client
{
public class BufferedFileRead
{
+ public ManualResetEvent WaitHandle;
+ public FileManager Manager;
+ public byte[] Buffer;
+ public int BufferOffset;
+ public long PieceStartIndex;
+ public int Count;
+
+
+ public BufferedFileRead(FileManager manager, byte[] buffer, int
bufferOffset, long pieceStartIndex, int count)
+ {
+ this.Manager = manager;
+ this.Buffer = buffer;
+ this.BufferOffset = bufferOffset;
+ this.PieceStartIndex = pieceStartIndex;
+ this.Count = count;
+ this.WaitHandle = new ManualResetEvent(false);
+ }
}
}
Modified: trunk/bitsharp/src/MonoTorrent/Client/BufferedFileWrite.cs
===================================================================
--- trunk/bitsharp/src/MonoTorrent/Client/BufferedFileWrite.cs 2008-01-03
16:46:00 UTC (rev 92189)
+++ trunk/bitsharp/src/MonoTorrent/Client/BufferedFileWrite.cs 2008-01-03
16:58:11 UTC (rev 92190)
@@ -3,6 +3,7 @@
using System.Text;
using MonoTorrent.Client.PeerMessages;
using MonoTorrent.Common;
+using System.Threading;
namespace MonoTorrent.Client
{
@@ -14,6 +15,7 @@
public PeerIdInternal Id;
public IPeerMessageInternal Message;
public Piece Piece;
+ public ManualResetEvent WaitHandle;
#endregion Fields
@@ -21,11 +23,18 @@
#region Constructors
public BufferedIO(PeerIdInternal id, ArraySegment<byte> buffer,
IPeerMessageInternal message, Piece piece)
+ : this(id, buffer, message, piece, null)
{
+
+ }
+
+ public BufferedIO(PeerIdInternal id, ArraySegment<byte> buffer,
IPeerMessageInternal message, Piece piece, ManualResetEvent handle)
+ {
this.Id = id;
this.Buffer = buffer;
this.Message = message;
this.Piece = piece;
+ this.WaitHandle = handle;
}
#endregion Constructors
Modified: trunk/bitsharp/src/MonoTorrent/Client/Managers/DiskManager.cs
===================================================================
--- trunk/bitsharp/src/MonoTorrent/Client/Managers/DiskManager.cs
2008-01-03 16:46:00 UTC (rev 92189)
+++ trunk/bitsharp/src/MonoTorrent/Client/Managers/DiskManager.cs
2008-01-03 16:58:11 UTC (rev 92190)
@@ -13,6 +13,7 @@
{
#region Member Variables
+ Queue<BufferedFileRead> bufferedReads;
Queue<BufferedIO> bufferedWrites;
private ClientEngine engine;
private int maxOpenStreams;
@@ -29,7 +30,6 @@
#region Old Variables
- //private FileStream[] fileStreams; // The
filestreams used to read/write to the files on disk
private bool ioActive; // Used to
signal when the IO thread is running
private Thread ioThread; // The
dedicated thread used for reading/writing
private object queueLock; // Used to
synchronise access on the IO thread
@@ -96,6 +96,7 @@
public DiskManager(ClientEngine engine)
{
+ this.bufferedReads = new Queue<BufferedFileRead>();
this.bufferedWrites = new Queue<BufferedIO>();
this.engine = engine;
this.ioActive = true;
@@ -105,7 +106,7 @@
this.rateLimiter = new RateLimiter();
this.streamsLock = new ReaderWriterLock();
this.threadWait = new ManualResetEvent(false);
- this.MaxWriteRate = 50;
+ this.MaxReadRate = 1000;
this.ioThread.Start();
}
@@ -134,6 +135,17 @@
/// <summary>
+ ///
+ /// </summary>
+ /// <param name="torrentManager"></param>
+ internal void CloseFileStreams(TorrentManager torrentManager)
+ {
+#warning Buffer the Filestreams
+ return;
+ }
+
+
+ /// <summary>
/// Disposes all necessary objects
/// </summary>
internal void Dispose()
@@ -178,8 +190,11 @@
/// <param name="s"></param>
internal void FreeStream(Stream s)
{
- s.Dispose();
- Interlocked.Decrement(ref openStreams);
+ if (s != null)
+ {
+ s.Dispose();
+ Interlocked.Decrement(ref openStreams);
+ }
}
@@ -260,9 +275,10 @@
/// Performs the buffered read
/// </summary>
/// <param name="bufferedFileIO"></param>
- private void PerformRead(BufferedIO bufferedFileIO)
+ private void PerformRead(BufferedFileRead io)
{
- throw new Exception("The method or operation is not implemented.");
+ Read(io.Manager, io.Buffer, io.BufferOffset, io.PieceStartIndex,
io.Count);
+ io.WaitHandle.Set();
}
@@ -290,22 +306,13 @@
}
- /// <summary>
- /// Queues a read request to be completed asynchronously
- /// </summary>
- /// <param name="id">The peer which the write request is for</param>
- /// <param name="recieveBuffer">The buffer to read the data
into</param>
- /// <param name="message">The RequestMessage</param>
- /// <param name="piece"></param>
- internal void QueueRead(PeerIdInternal id, ArraySegment<byte>
recieveBuffer, RequestMessage message, Piece piece)
+ internal void QueueRead(BufferedFileRead io)
{
- throw new NotImplementedException();
-
- /*lock (this.queueLock)
+ lock (this.queueLock)
{
- this.bufferedReads.Enqueue(new BufferedIO(id, recieveBuffer,
message, piece));
+ bufferedReads.Enqueue(io);
SetHandleState(true);
- }*/
+ }
}
@@ -372,11 +379,11 @@
private void RunIO()
{
BufferedIO write;
- //BufferedIO read;
+ BufferedFileRead read;
while (ioActive)
{
write = null;
- //read = null;
+ read = null;
// Take a lock on the queue and dequeue any reads/writes that
are available. Then lose the lock before
// performing the actual read/write to avoid blocking other
threads
@@ -387,19 +394,26 @@
write = this.bufferedWrites.Dequeue();
Interlocked.Add(ref rateLimiter.DownloadChunks,
-write.Buffer.Count / ConnectionManager.ChunkLength);
}
- //if (this.bufferedReads.Count > 0)
- // read = this.bufferedReads.Dequeue();
- // If there are no more reads available and no more writes
available, set the handle to wait
- if (this.bufferedWrites.Count == 0 || write == null) //
this.bufferedReads.Count == 0 &&
+ if (this.bufferedReads.Count > 0 &&
(rateLimiter.UploadChunks > 0))
+ {
+ read = this.bufferedReads.Dequeue();
+ Interlocked.Add(ref rateLimiter.UploadChunks,
-read.Count / ConnectionManager.ChunkLength);
+ }
+
+ // If both the read queue and write queue are empty, then
we unset the handle.
+ // Or if we have reached the max read/write rate and can't
dequeue something, we unset the handle
+ if ((this.bufferedWrites.Count == 0 &&
this.bufferedReads.Count == 0) || (write == null) && (read == null))
SetHandleState(false);
}
if (write != null)
PerformWrite(write);
- //if (read != null)
- // PerformRead(read);
+ if (read != null)
+ PerformRead(read);
+
+ // Wait ~100 ms before trying to read/write something again to
give the rate limiting a chance to recover
this.threadWait.WaitOne(100, false);
}
}
@@ -487,10 +501,5 @@
#endregion
- internal void CloseFileStreams(TorrentManager torrentManager)
- {
-#warning Buffer the Filestreams
- return;
- }
}
}
Modified: trunk/bitsharp/src/MonoTorrent/Client/Managers/FileManager.cs
===================================================================
--- trunk/bitsharp/src/MonoTorrent/Client/Managers/FileManager.cs
2008-01-03 16:46:00 UTC (rev 92189)
+++ trunk/bitsharp/src/MonoTorrent/Client/Managers/FileManager.cs
2008-01-03 16:58:11 UTC (rev 92190)
@@ -214,8 +214,10 @@
if ((pieceStartIndex + bytesToRead) > this.fileSize)
bytesToRead -= (int)((pieceStartIndex +
bytesToRead) - fileSize);
- //using (new ReaderLock(this.streamsLock))
- bytesRead = manager.Engine.DiskManager.Read(this,
hashBuffer.Array, hashBuffer.Offset, pieceStartIndex, bytesToRead);
+ BufferedFileRead read = new BufferedFileRead(this,
hashBuffer.Array, hashBuffer.Offset, pieceStartIndex, bytesToRead);
+ manager.Engine.DiskManager.QueueRead(read);
+ read.WaitHandle.WaitOne();
+ read.WaitHandle.Close();
hasher.TransformBlock(hashBuffer.Array,
hashBuffer.Offset, bytesRead, hashBuffer.Array, hashBuffer.Offset);
totalRead += bytesRead;
Modified: trunk/bitsharp/src/MonoTorrent/Client/Managers/TorrentManager.cs
===================================================================
--- trunk/bitsharp/src/MonoTorrent/Client/Managers/TorrentManager.cs
2008-01-03 16:46:00 UTC (rev 92189)
+++ trunk/bitsharp/src/MonoTorrent/Client/Managers/TorrentManager.cs
2008-01-03 16:58:11 UTC (rev 92190)
@@ -384,6 +384,7 @@
/// </summary>
public void Start()
{
+ this.engine.Start();
lock (this.engine.asyncCompletionLock)
{
// If the torrent was "paused", then just update the state to
Downloading and forcefully
@@ -422,7 +423,6 @@
engine.ConnectionManager.RegisterManager(this);
this.pieceManager.Reset();
- this.engine.Start();
}
}
_______________________________________________
Mono-patches maillist - [email protected]
http://lists.ximian.com/mailman/listinfo/mono-patches
