Commit c9c1b040 authored by Marc Gravell's avatar Marc Gravell

Report available inbound bytes on socket

parent 6718f8a9
...@@ -1702,13 +1702,15 @@ internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, Ser ...@@ -1702,13 +1702,15 @@ internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, Ser
} }
else else
{ {
int inst, qu, qs, qc, wr, wq; int inst, qu, qs, qc, wr, wq, @in;
int queue = server.GetOutstandingCount(message.Command, out inst, out qu, out qs, out qc, out wr, out wq); int queue = server.GetOutstandingCount(message.Command, out inst, out qu, out qs, out qc, out wr, out wq, out @in);
var sb = new StringBuilder("Timeout performing ").Append(message.CommandAndKey) var sb = new StringBuilder("Timeout performing ").Append(message.CommandAndKey)
.Append(", inst: ").Append(inst) .Append(", inst: ").Append(inst)
.Append(", queue: ").Append(queue).Append(", qu=").Append(qu) .Append(", queue: ").Append(queue).Append(", qu=").Append(qu)
.Append(", qs=").Append(qs).Append(", qc=").Append(qc) .Append(", qs=").Append(qs).Append(", qc=").Append(qc)
.Append(", wr=").Append(wr).Append("/").Append(wq); .Append(", wr=").Append(wr).Append("/").Append(wq)
.Append(", in=").Append(@in);
errMessage = sb.ToString(); errMessage = sb.ToString();
if (stormLogThreshold >= 0 && queue >= stormLogThreshold && Interlocked.CompareExchange(ref haveStormLog, 1, 0) == 0) if (stormLogThreshold >= 0 && queue >= stormLogThreshold && Interlocked.CompareExchange(ref haveStormLog, 1, 0) == 0)
{ {
......
...@@ -209,12 +209,19 @@ internal void GetCounters(ConnectionCounters counters) ...@@ -209,12 +209,19 @@ internal void GetCounters(ConnectionCounters counters)
} }
} }
internal int GetOutstandingCount(out int inst, out int qu, out int qs, out int qc, out int wr, out int wq) internal int GetOutstandingCount(out int inst, out int qu, out int qs, out int qc, out int wr, out int wq, out int @in)
{// defined as: PendingUnsentItems + SentItemsAwaitingResponse + ResponsesAwaitingAsyncCompletion {// defined as: PendingUnsentItems + SentItemsAwaitingResponse + ResponsesAwaitingAsyncCompletion
inst = (int)(Interlocked.Read(ref operationCount) - Interlocked.Read(ref profileLastLog)); inst = (int)(Interlocked.Read(ref operationCount) - Interlocked.Read(ref profileLastLog));
qu = queue.Count(); qu = queue.Count();
var tmp = physical; var tmp = physical;
qs = tmp == null ? 0 : tmp.GetSentAwaitingResponseCount(); if(tmp == null)
{
qs = @in = 0;
} else
{
qs = tmp.GetSentAwaitingResponseCount();
@in = tmp.GetAvailableInboundBytes();
}
qc = completionManager.GetOutstandingCount(); qc = completionManager.GetOutstandingCount();
wr = Interlocked.CompareExchange(ref activeWriters, 0, 0); wr = Interlocked.CompareExchange(ref activeWriters, 0, 0);
wq = Interlocked.CompareExchange(ref inWriteQueue, 0, 0); wq = Interlocked.CompareExchange(ref inWriteQueue, 0, 0);
......
...@@ -513,6 +513,11 @@ void BeginReading() ...@@ -513,6 +513,11 @@ void BeginReading()
} while (keepReading); } while (keepReading);
} }
internal int GetAvailableInboundBytes()
{
return this.socketToken.Available;
}
SocketMode ISocketCallback.Connected(Stream stream) SocketMode ISocketCallback.Connected(Stream stream)
{ {
try try
......
...@@ -329,15 +329,15 @@ internal ServerCounters GetCounters() ...@@ -329,15 +329,15 @@ internal ServerCounters GetCounters()
return counters; return counters;
} }
internal int GetOutstandingCount(RedisCommand command, out int inst, out int qu, out int qs, out int qc, out int wr, out int wq) internal int GetOutstandingCount(RedisCommand command, out int inst, out int qu, out int qs, out int qc, out int wr, out int wq, out int @in)
{ {
var bridge = GetBridge(command, false); var bridge = GetBridge(command, false);
if (bridge == null) if (bridge == null)
{ {
return inst = qu = qs = qc = wr = wq = 0; return inst = qu = qs = qc = wr = wq = @in = 0;
} }
return bridge.GetOutstandingCount(out inst, out qu, out qs, out qc, out wr, out wq); return bridge.GetOutstandingCount(out inst, out qu, out qs, out qc, out wr, out wq, out @in);
} }
internal string GetProfile() internal string GetProfile()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment