Availability Anywhere Part 13 — Optimizing FileProxy

This is the thirteenth part of the Availability Anywhere series. For your convenience you can find other parts in the table of contents in Part 1 – Connecting to SSH tunnel automatically in Windows

Last time we implemented TCP over File System (FileProxy). Today we’re going to check its performance and optimize it a bit.

Let’s start with benchmarks. We are going to use the following application:

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Runtime.Remoting.Channels;
using System.Text;
using System.Threading;

namespace EchoBenchmark
{
    class Program
    {
        static void Main(string[] args)
        {
            if (args.Length < 2)
            {
                Exit(Usage());
            }

            if (args[0] == "client")
            {
                Benchmark(args);
            }
            else if (args[0] == "server")
            {
                Server(args);
            }
            else
            {
                Exit(Usage());
            }
        }

        private static void Benchmark(string[] args)
        {
            Console.WriteLine("Mode: " + args[2]);

            int maxSize = 100 * 1024;
            int attempts = 1000;
            var stopwatch = new Stopwatch();

            int destinationPort = int.Parse(args[1]);
            for (int size = 1; size <= maxSize; size *= 10)
            {
                IPEndPoint remoteEP = new IPEndPoint(Dns.GetHostEntry("localhost").AddressList.First(a => a.AddressFamily == AddressFamily.InterNetwork), destinationPort);
                Socket senderSocket = new Socket(remoteEP.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

                senderSocket.Connect(remoteEP);

                var sendBuffer = new byte[size];
                var receiveBuffer = new byte[size];

                stopwatch.Restart();
                for (int attempt = 0; attempt < attempts; ++attempt)
                {
                    senderSocket.Send(sendBuffer);

                    int totalRead = 0;
                    while (totalRead < size)
                    {
                        totalRead += senderSocket.Receive(receiveBuffer);
                    }
                }

                stopwatch.Stop();
                Console.WriteLine(size + "\t" + (double)stopwatch.ElapsedMilliseconds / attempts);

                senderSocket.Close();
            }
        }

        private static void Server(string[] args)
        {
            int localPort = int.Parse(args[1]);
            IPEndPoint localEndPoint = new IPEndPoint(0, localPort);
            Socket listener = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            listener.Bind(localEndPoint);
            listener.Listen(100);

            while (true)
            {
                Socket socket = listener.Accept();
                Console.WriteLine("New connection accepted to be scattered");

                new Thread(() =>
                {
                    while (true)
                    {
                        var buffer = new byte[100000];
                        while (true)
                        {
                            var read = socket.Receive(buffer);

                            if (read == 0)
                            {
                                return;
                            }

                            socket.Send(buffer, 0, read, SocketFlags.None);
                        }
                    }
                }).Start();
            }
        }

        private static string Usage()
        {
            return "EchoBenchmark.exe client local_port\nEchoBenchmark.exe server local_port";
        }

        private static void Exit(string message)
        {
            Console.WriteLine(message);
            Environment.Exit(0);
        }
    }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

using System;

using System.Collections.Generic;

using System.Diagnostics;

using System.Linq;

using System.Net;

using System.Net.Sockets;

using System.Runtime.Remoting.Channels;

using System.Text;

using System.Threading;

namespace EchoBenchmark

{

class Program

{

static void Main(string[] args)

{

if (args.Length < 2)

{

Exit(Usage());

}

if (args[0] == "client")

{

Benchmark(args);

}

else if (args[0] == "server")

{

Server(args);

}

else

{

Exit(Usage());

}

private static void Benchmark(string[] args)

{

Console.WriteLine("Mode: " + args[2]);

int maxSize = 100 * 1024;

int attempts = 1000;

var stopwatch = new Stopwatch();

int destinationPort = int.Parse(args[1]);

for (int size = 1; size <= maxSize; size *= 10)

{

IPEndPoint remoteEP = new IPEndPoint(Dns.GetHostEntry("localhost").AddressList.First(a => a.AddressFamily == AddressFamily.InterNetwork), destinationPort);

Socket senderSocket = new Socket(remoteEP.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

senderSocket.Connect(remoteEP);

var sendBuffer = new byte[size];

var receiveBuffer = new byte[size];

stopwatch.Restart();

for (int attempt = 0; attempt < attempts; ++attempt)

{

senderSocket.Send(sendBuffer);

int totalRead = 0;

while (totalRead < size)

{

totalRead += senderSocket.Receive(receiveBuffer);

}

stopwatch.Stop();

Console.WriteLine(size + "\t" + (double)stopwatch.ElapsedMilliseconds / attempts);

senderSocket.Close();

}

private static void Server(string[] args)

{

int localPort = int.Parse(args[1]);

IPEndPoint localEndPoint = new IPEndPoint(0, localPort);

Socket listener = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

listener.Bind(localEndPoint);

listener.Listen(100);

while (true)

{

Socket socket = listener.Accept();

Console.WriteLine("New connection accepted to be scattered");

new Thread(() =>

{

while (true)

{

var buffer = new byte[100000];

while (true)

{

var read = socket.Receive(buffer);

if (read == 0)

{

return;

}

socket.Send(buffer, 0, read, SocketFlags.None);

}

}).Start();

}

private static string Usage()

{

return "EchoBenchmark.exe client local_port\nEchoBenchmark.exe server local_port";

}

private static void Exit(string message)

{

Console.WriteLine(message);

Environment.Exit(0);

}

This application has two modes. In the client mode it generates some traffic to the server. In the server mode it reads the incoming traffic and echoes it back.

The client sends packets of sizes between 1 byte and 100 000 bytes. It sends every packet one thousand times and then calculates how long it took on average to send one packet.

We run this application twice. First, with direct connection between the client and the server. Second time with the TCP over File System proxy in between. Here are the results in milliseconds:

Mode: Direct
1       0.051
10      0.048
100     0.039
1000    0.033
10000   0.079
100000  0.697

Mode: Direct

1 0.051

10 0.048

100 0.039

1000 0.033

10000 0.079

100000 0.697

Mode: FileProxy
1       3.463
10      3.019
100     3.075
1000    3.004
10000   2.771
100000  5.523

Mode: FileProxy

1 3.463

10 3.019

100 3.075

1000 3.004

10000 2.771

100000 5.523

So we can see that the direct connection has an overhead of 40-700 microseconds. FileProxy makes it between 3 and 5 milliseconds. So the total slowdown is up to 70 times. However, this is even slower when run between the host and the guest VM:

Mode: FileProxyVirtualMachine
1       249.961
10      28.835
100     86.738
1000    80.495
10000   95.026
100000  101.823

Mode: FileProxyVirtualMachine

1 249.961

10 28.835

100 86.738

1000 80.495

10000 95.026

100000 101.823

This time it is even 5000 times slower! However, average overhead is around 80-100 milliseconds. This makes interactive applications much slower.

Let’s see what we do wrong. We don’t handle permissions to the file well, so two processes throw exceptions when the file is blocked. We try reading the file constantly instead of getting notified when the file is changed. We resize arrays instead of sending the buffer directly. Here is the updated code:

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;

namespace FileProxy
{
    class Program
    {
        static void Main(string[] args)
        {
            if (args.Length < 2)
            {
                Exit(Usage());
            }

            if (args[0] == "client")
            {
                Client.Start(args);
            }
            else if (args[0] == "server")
            {
                Server.Start(args);
            }
            else
            {
                Exit(Usage());
            }
        }

        private static string Usage()
        {
            return "FileProxy.exe client local_port client_directory server_directory\nFileProxy.exe server destination_ip:destination_port client_directory server_directory";
        }

        private static void Exit(string message)
        {
            Console.WriteLine(message);
            Environment.Exit(0);
        }
    }
	
    class Server
    {
        public static void Start(string[] args)
        {
            var destinationIp = args[1].Split(':')[0];
            var destinationPort = int.Parse(args[1].Split(':')[1]);
            var clientDirectory = args[2];
            var serverDirectory = args[3];

            Console.WriteLine($"Routing to {destinationIp}:{destinationPort} via {clientDirectory}<=>{serverDirectory}");

            var existingFiles = new HashSet<string>();

            while (true)
            {
                try
                {
                    while (true)
                    {
                        foreach (var file in Directory.EnumerateFiles(serverDirectory).Select(Path.GetFileName))
                        {
                            if (existingFiles.Add(file))
                            {
                                Console.WriteLine("New connection accepted to be scattered: " + file);

                                // I'm forcing IPv4 (IPv6 breaks with Cisco AnyConnect)
                                IPEndPoint remoteEP = new IPEndPoint(Dns.GetHostEntry(destinationIp).AddressList.First(a => a.AddressFamily == AddressFamily.InterNetwork), destinationPort);
                                Socket senderSocket = new Socket(remoteEP.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

                                senderSocket.Connect(remoteEP);
                                Console.WriteLine("Socket connected to {0}", senderSocket.RemoteEndPoint);

                                new Thread(() =>
                                {

                                    var socket = new FileSocket(senderSocket, int.Parse(file), serverDirectory,
                                        clientDirectory, r => { }, s => { });
                                    socket.Start();
                                }).Start();
                            }
                        }

                        Thread.Sleep(1000);
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Exception " + e);
                }
            }
        }
    }
	
    public class Client
    {
        private static Random random = new Random();

        public static void Start(string[] args)
        {
            var localPort = int.Parse(args[1]);
            var clientDirectory = args[2];
            var serverDirectory = args[3];

            Console.WriteLine($"Routing from {localPort} via {clientDirectory}<=>{serverDirectory}");

            IPEndPoint localEndPoint = new IPEndPoint(0, localPort);
            Socket listener = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            listener.Bind(localEndPoint);
            listener.Listen(100);

            var totalSent = new Dictionary<string, long>();
            var totalReceived = new Dictionary<string, long>();
            var totalExceptions = new Dictionary<string, long>();
            var localDnss = new[] {"localhost"};


            foreach (var localDns in localDnss)
            {
                totalSent[localDns] = 0;
                totalReceived[localDns] = 0;
                totalExceptions[localDns] = 0;
            }

            new Thread(() =>
            {
                while (true)
                {
                    Console.Write(DateTime.Now);
                    Console.Write(" E/S/R:\t");
                    Console.WriteLine(string.Join("\t", localDnss.Select(dns => $"{dns}: {totalExceptions[dns]}/{totalSent[dns]}/{totalReceived[dns]}")));

                    Thread.Sleep(3000);
                }
            }).Start();

            while (true)
            {
                try
                {
                    while (true)
                    {
                        Socket socket = listener.Accept();
                        Console.WriteLine("New connection accepted to be scattered");

                        new Thread(() => Start(socket, localDnss, clientDirectory, serverDirectory, random.Next(), totalSent, totalReceived, totalExceptions)).Start();
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Exception: " + e);
                }
            }
        }

        private static void Start(Socket clientSocket,string[] localDnss,  string clientDirectory, string serverDirectory, int identifier,
            Dictionary<string, long> sent, Dictionary<string, long> received, Dictionary<string, long> exceptions)
        {
            try
            {
                var fileSocket = new FileSocket(clientSocket, identifier, clientDirectory, serverDirectory, r => received[localDnss[0]]+= r, s => sent[localDnss[0]]+=s);
                fileSocket.Start();
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception: " + e);
            }
        }
    }
	
    public class FileSocket
    {
        private Socket clientSocket;
        private int identifier;
        string readingDirectory;
        private string writingDirectory;
        private Action<int> received;
        private Action<int> sent;
        public FileSocket(Socket socket, int identifier, string readingDirectory, string writingDirectory, Action<int> received, Action<int> sent)
        {
            this.clientSocket = socket;
            this.identifier = identifier;
            this.readingDirectory = readingDirectory;
            this.writingDirectory = writingDirectory;
            this.received = received;
            this.sent = sent;
        }

        public void Start()
        {
            Thread clientThread = new Thread(() =>
            {
                try
                {
                    var buffer = new byte[100000];
                    var path = Path.Combine(writingDirectory, identifier + "");
                    while (true)
                    {
                        var read = clientSocket.Receive(buffer);

                        if (read == 0)
                        {
                            throw new Exception("Socket was closed");
                        }

                        while (true)
                        {
                            try
                            {
                                using (var stream = new FileStream(path, FileMode.Append, FileAccess.Write, FileShare.ReadWrite))
                                {
                                    stream.Write(buffer, 0, read);
                                    sent(read);
                                    break;
                                }
                            }
                            catch (Exception e2)
                            {
                                Console.WriteLine("Exception when writing to file" + e2);
                            }
                        }
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Exception " + e);
                    clientSocket.Close();
                }
            });

            try
            {
                Thread senderThread = new Thread(KeepReading);
                senderThread.Start();
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception: " + e);
            }

            clientThread.Start();
            clientThread.Join();
        }

        private void KeepReading()
        {
            int totalRead = 0;
            int toRead = 100024;
            var locker = new ManualResetEventSlim();

            var watcher = new FileSystemWatcher(readingDirectory);
            watcher.NotifyFilter = NotifyFilters.LastWrite
                                   | NotifyFilters.Size;

            watcher.Changed += (e, s) => { locker.Set(); };
            watcher.Error += (e, s) => { locker.Set(); };
            watcher.EnableRaisingEvents = true;

            byte[] bytes = new byte[toRead];
            while (true)
            {
                try
                {
                    using (FileStream fileStream = new FileStream(Path.Combine(readingDirectory, identifier + ""), FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
                    {
                        while (true)
                        {
                            fileStream.Seek(totalRead, SeekOrigin.Begin);
                            int howMuchRead = fileStream.Read(bytes, 0, toRead);

                            if (howMuchRead == 0)
                                break;

                            totalRead += howMuchRead;
                            received(howMuchRead);
                            clientSocket.Send(bytes, 0, howMuchRead, SocketFlags.None);
                        }
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Exception while reading: " + e);
                }

                locker.Wait();
                locker.Reset();
            }
        }
    }
}
</pre.

Results:

<pre>
Mode: FileProxy Optimized
1       1.388
10      0.971
100     1.034
1000    1.11
10000   1.313
100000  4.175
</pre.

<pre>
Mode: FileProxyVirtualMachine Optimized
1       7.629
10      9.26
100     14.59
1000    16.796
10000   24.379
100000  27.509

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

using System;

using System.Collections.Generic;

using System.IO;

using System.Linq;

using System.Net;

using System.Net.Sockets;

using System.Threading;

namespace FileProxy

{

class Program

{

static void Main(string[] args)

{

if (args.Length < 2)

{

Exit(Usage());

}

if (args[0] == "client")

{

Client.Start(args);

}

else if (args[0] == "server")

{

Server.Start(args);

}

else

{

Exit(Usage());

}

private static string Usage()

{

return "FileProxy.exe client local_port client_directory server_directory\nFileProxy.exe server destination_ip:destination_port client_directory server_directory";

}

private static void Exit(string message)

{

Console.WriteLine(message);

Environment.Exit(0);

}

class Server

{

public static void Start(string[] args)

{

var destinationIp = args[1].Split(':')[0];

var destinationPort = int.Parse(args[1].Split(':')[1]);

var clientDirectory = args[2];

var serverDirectory = args[3];

Console.WriteLine($"Routing to {destinationIp}:{destinationPort} via {clientDirectory}<=>{serverDirectory}");

var existingFiles = new HashSet<string>();

while (true)

{

try

{

while (true)

{

foreach (var file in Directory.EnumerateFiles(serverDirectory).Select(Path.GetFileName))

{

if (existingFiles.Add(file))

{

Console.WriteLine("New connection accepted to be scattered: " + file);

// I'm forcing IPv4 (IPv6 breaks with Cisco AnyConnect)

IPEndPoint remoteEP = new IPEndPoint(Dns.GetHostEntry(destinationIp).AddressList.First(a => a.AddressFamily == AddressFamily.InterNetwork), destinationPort);

Socket senderSocket = new Socket(remoteEP.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

senderSocket.Connect(remoteEP);

Console.WriteLine("Socket connected to {0}", senderSocket.RemoteEndPoint);

new Thread(() =>

{

var socket = new FileSocket(senderSocket, int.Parse(file), serverDirectory,

clientDirectory, r => { }, s => { });

socket.Start();

}).Start();

}

Thread.Sleep(1000);

}

catch (Exception e)

{

Console.WriteLine("Exception " + e);

}

public class Client

{

private static Random random = new Random();

public static void Start(string[] args)

{

var localPort = int.Parse(args[1]);

var clientDirectory = args[2];

var serverDirectory = args[3];

Console.WriteLine($"Routing from {localPort} via {clientDirectory}<=>{serverDirectory}");

IPEndPoint localEndPoint = new IPEndPoint(0, localPort);

Socket listener = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

listener.Bind(localEndPoint);

listener.Listen(100);

var totalSent = new Dictionary<string, long>();

var totalReceived = new Dictionary<string, long>();

var totalExceptions = new Dictionary<string, long>();

var localDnss = new[] {"localhost"};

foreach (var localDns in localDnss)

{

totalSent[localDns] = 0;

totalReceived[localDns] = 0;

totalExceptions[localDns] = 0;

}

new Thread(() =>

{

while (true)

{

Console.Write(DateTime.Now);

Console.Write(" E/S/R:\t");

Console.WriteLine(string.Join("\t", localDnss.Select(dns => $"{dns}: {totalExceptions[dns]}/{totalSent[dns]}/{totalReceived[dns]}")));

Thread.Sleep(3000);

}

}).Start();

while (true)

{

try

{

while (true)

{

Socket socket = listener.Accept();

Console.WriteLine("New connection accepted to be scattered");

new Thread(() => Start(socket, localDnss, clientDirectory, serverDirectory, random.Next(), totalSent, totalReceived, totalExceptions)).Start();

}

catch (Exception e)

{

Console.WriteLine("Exception: " + e);

}

private static void Start(Socket clientSocket,string[] localDnss, string clientDirectory, string serverDirectory, int identifier,

Dictionary<string, long> sent, Dictionary<string, long> received, Dictionary<string, long> exceptions)

{

try

{

var fileSocket = new FileSocket(clientSocket, identifier, clientDirectory, serverDirectory, r => received[localDnss[0]]+= r, s => sent[localDnss[0]]+=s);

fileSocket.Start();

}

catch (Exception e)

{

Console.WriteLine("Exception: " + e);

}

public class FileSocket

{

private Socket clientSocket;

private int identifier;

string readingDirectory;

private string writingDirectory;

private Action<int> received;

private Action<int> sent;

public FileSocket(Socket socket, int identifier, string readingDirectory, string writingDirectory, Action<int> received, Action<int> sent)

{

this.clientSocket = socket;

this.identifier = identifier;

this.readingDirectory = readingDirectory;

this.writingDirectory = writingDirectory;

this.received = received;

this.sent = sent;

}

public void Start()

{

Thread clientThread = new Thread(() =>

{

try

{

var buffer = new byte[100000];

var path = Path.Combine(writingDirectory, identifier + "");

while (true)

{

var read = clientSocket.Receive(buffer);

if (read == 0)

{

throw new Exception("Socket was closed");

}

while (true)

{

try

{

using (var stream = new FileStream(path, FileMode.Append, FileAccess.Write, FileShare.ReadWrite))

{

stream.Write(buffer, 0, read);

sent(read);

break;

}

catch (Exception e2)

{

Console.WriteLine("Exception when writing to file" + e2);

}

catch (Exception e)

{

Console.WriteLine("Exception " + e);

clientSocket.Close();

}

});

try

{

Thread senderThread = new Thread(KeepReading);

senderThread.Start();

}

catch (Exception e)

{

Console.WriteLine("Exception: " + e);

}

clientThread.Start();

clientThread.Join();

}

private void KeepReading()

{

int totalRead = 0;

int toRead = 100024;

var locker = new ManualResetEventSlim();

var watcher = new FileSystemWatcher(readingDirectory);

watcher.NotifyFilter = NotifyFilters.LastWrite

| NotifyFilters.Size;

watcher.Changed += (e, s) => { locker.Set(); };

watcher.Error += (e, s) => { locker.Set(); };

watcher.EnableRaisingEvents = true;

byte[] bytes = new byte[toRead];

while (true)

{

try

{

using (FileStream fileStream = new FileStream(Path.Combine(readingDirectory, identifier + ""), FileMode.Open, FileAccess.Read, FileShare.ReadWrite))

{

while (true)

{

fileStream.Seek(totalRead, SeekOrigin.Begin);

int howMuchRead = fileStream.Read(bytes, 0, toRead);

if (howMuchRead == 0)

break;

totalRead += howMuchRead;

received(howMuchRead);

clientSocket.Send(bytes, 0, howMuchRead, SocketFlags.None);

}

catch (Exception e)

{

Console.WriteLine("Exception while reading: " + e);

}

locker.Wait();

locker.Reset();

}

</pre.

Results:

<pre>

Mode: FileProxy Optimized

1 1.388

10 0.971

100 1.034

1000 1.11

10000 1.313

100000 4.175

</pre.

<pre>

Mode: FileProxyVirtualMachine Optimized

1 7.629

10 9.26

100 14.59

1000 16.796

10000 24.379

100000 27.509

We can see that we optimized the application 3-4 times. This gives a much better performance.

What makes the slowdown between the host and the guest? The filesystem that is not shared well. We could try using RAM Disk based on imDisk:

Mode: FileProxy Optimized with RAM Disk
1       1.938
10      3.899
100     9.814
1000    10.716
10000   10.334
100000  13.052

Mode: FileProxy Optimized with RAM Disk

1 1.938

10 3.899

100 9.814

1000 10.716

10000 10.334

100000 13.052

Mode: FileProxyVirtualMachine Optimized with RAM Disk
1       11.222
10      21.506
100     30.05
1000    32.314
10000   28.891
100000  52.593

Mode: FileProxyVirtualMachine Optimized with RAM Disk

1 11.222

10 21.506

100 30.05

1000 32.314

10000 28.891

100000 52.593

We can see that it’s actually slower. I don’t know if there is a way to make this much aster. Obiviously, we could optimize the code a little bit more, but I think the drive is the biggest factor here.