The Ecere SDK - sdk/blob - ecere/src/net/NetworkClientFile.ec

   1 namespace net;
   2
   3 #ifndef ECERE_NONET
   4
   5 import "network"
   6
   7 #define PUTXDWORD(b, d) \
   8    (b)[0] = (byte)(((d) >> 24) & 0xFF); \
   9    (b)[1] = (byte)(((d) >> 16) & 0xFF); \
  10    (b)[2] = (byte)(((d) >> 8)  & 0xFF); \
  11    (b)[3] = (byte)( (d)        & 0xFF);
  12 #define XDWORD(d) \
  13    { \
  14       (byte)(((d) >> 24) & 0xFF), \
  15       (byte)(((d) >> 16) & 0xFF), \
  16       (byte)(((d) >> 8)  & 0xFF), \
  17       (byte)((d)         & 0xFF) \
  18    }
  19 #define GETXDWORD(b) (uint32)(((b)[0] << 24) | ((b)[1] << 16) | ((b)[2] << 8) | (b)[3])
  20
  21 /*
  22 #define PACKET_SIZE     6 * 1024 * 1024 //65536 //16384
  23 #define MAXBUFFERSIZE   6 * 1024 * 1024 //104867572
  24 #define BUFFER_SIZE     6 * 1024 * 1024 //65536 * 8 //16384 * 8 //5001265536
  25 */
  26
  27 #define MAXBUFFERSIZE   1024*1024
  28 #define PACKET_SIZE     65530 //131072
  29 #define BUFFER_SIZE     PACKET_SIZE * 8
  30
  31 // Packets sent from Client to Server
  32 static struct NetworkClientPacket
  33 {
  34    byte type[4];
  35    byte size[4];
  36 };
  37
  38 static class OpenPacket : NetworkClientPacket
  39 {
  40    byte fileNameLen[4];
  41    char fileName[1]; // Rest follows, null terminated
  42 };
  43
  44 static struct ClosePacket : NetworkClientPacket
  45 {
  46    byte handle[4];
  47 };
  48
  49 static struct ReadPacket : NetworkClientPacket
  50 {
  51    byte handle[4];
  52    byte readStart[4];
  53    byte readSize[4];
  54 };
  55
  56 static struct SizePacket : NetworkClientPacket
  57 {
  58    byte handle[4];
  59 };
  60
  61 static class NetDataBlock : struct
  62 {
  63    NetDataBlock prev, next;
  64    uint start, end;
  65 };
  66
  67 static enum NetworkClientPacketType { open, close, read, getSize };
  68
  69 static class Request : struct
  70 {
  71    Request prev, next;
  72    uint numBytes;
  73    void (* CallBack)(Request request, const byte * buffer);
  74
  75    NetworkClientFile file;
  76    Semaphore semaphore;
  77    uint readStart, readSize;
  78
  79    void ReadCallback(const byte * buffer)
  80    {
  81       NetworkClientFile file = this.file;
  82       FileServerConnection connection = file.connection;
  83       connection.mutex.Wait();
  84       {
  85          // Check where this goes in the buffer
  86          int overlap = file.position + file.bufferCount - (int)readStart;
  87          int bytesToRead = numBytes - (int)overlap;
  88
  89          // Make an exception if this is an actual read (it's been waited on...)
  90          if(!semaphore)
  91             bytesToRead = Min(bytesToRead, MAXBUFFERSIZE - (int)file.bufferCount);
  92
  93          if(bytesToRead > 0)
  94          {
  95             if(file.bufferSize < file.bufferCount + bytesToRead)
  96             {
  97                file.buffer = renew file.buffer byte[file.bufferCount + bytesToRead];
  98                file.bufferSize = file.bufferCount + bytesToRead;
  99             }
 100             memcpy(file.buffer + file.bufferCount, buffer + overlap, bytesToRead);
 101             file.bufferCount += bytesToRead;
 102          }
 103
 104          // Request some more data!
 105          if(file.bufferCount < BUFFER_SIZE)
 106          {
 107             // Don't do it if we know it's passed end of file
 108             if(numBytes)
 109                file.ReadMoreData(file.position + file.bufferCount, PACKET_SIZE, false);
 110          }
 111
 112          if(semaphore)
 113             // This request is being waited on... we can't free it yet
 114             semaphore.Release();
 115          else
 116             delete this;
 117       }
 118       connection.mutex.Release();
 119    }
 120
 121    void ReadResultCallback(const byte * buffer)
 122    {
 123       NetworkClientFile file = this.file;
 124       file.connection.mutex.Wait();
 125       // Reuse this request, reinsert it at the head
 126       file.connection.requests.Insert(null, this);
 127       numBytes = GETXDWORD(buffer);
 128       CallBack = ReadCallback;
 129
 130       file.connection.mutex.Release();
 131    }
 132
 133    void SizeCallback(const byte * buffer)
 134    {
 135       readSize = GETXDWORD(buffer);
 136       semaphore.Release();
 137    }
 138
 139    void OpenCallback(const byte * buffer)
 140    {
 141       file.handle = GETXDWORD(buffer);
 142       semaphore.Release();
 143    }
 144 };
 145
 146 class FileClientThread : Thread
 147 {
 148    FileServerConnection socket;
 149    uint Main()
 150    {
 151       while(!terminate)
 152       {
 153          //guiApp.WaitNetworkEvent();
 154          //guiApp.ProcessNetworkEvents();
 155          socket.Process();
 156       }
 157       return 0;
 158    }
 159    bool terminate;
 160 }
 161
 162 public class NetworkClientFile : File
 163 {
 164    // The only guys who will be moving where the buffer data is mapped to the file are Read or Seek!
 165    int Read(byte * buffer, uint size, uint count)
 166    {
 167       FileServerConnection connection = this.connection;
 168       uint bytesRead = 0;
 169
 170       count *= size;
 171
 172       connection.mutex.Wait();
 173
 174       // buffer is always data at virtual position in the file
 175       // position is virtual position in the file
 176       bytesRead = bufferCount;
 177       if(bytesRead < count)
 178       {
 179          connection.mutex.Release();
 180          // Here we are missing (count - bytesRead) bytes starting at file->position
 181          // ReadMoreData(position + bufferCount, (count - bufferCount), true);
 182          ReadMoreData(position + bufferCount, (Max(count, BUFFER_SIZE) - bufferCount), true);
 183          connection.mutex.Wait();
 184       }
 185       bytesRead = Min(count, bufferCount);
 186
 187       memcpy(buffer, this.buffer, bytesRead);
 188       memcpy(this.buffer, this.buffer + bytesRead, bufferCount - bytesRead);
 189       bufferCount -= bytesRead;
 190       position += bytesRead;
 191
 192       connection.mutex.Release();
 193       return bytesRead;
 194    }
 195
 196    int Write(byte * buffer, uint size, uint count)
 197    {
 198       return 0;
 199    }
 200
 201    bool Getc(char * ch)
 202    {
 203       return 0;
 204    }
 205
 206    bool Putc(char ch)
 207    {
 208       return 0;
 209    }
 210
 211    bool Puts(char * string)
 212    {
 213       return 0;
 214    }
 215
 216    bool Seek(int pos, FileSeekMode mode)
 217    {
 218       uint newPosition = position;
 219       switch(mode)
 220       {
 221          case start:
 222             newPosition = pos;
 223             break;
 224          case current:
 225             newPosition += pos;
 226             break;
 227          case end:
 228          {
 229             // Get size of file
 230             connection.mutex.Wait();
 231             {
 232                Request request = connection.AddRequest(4, Request::SizeCallback, this, true);
 233                connection.SendSizePacket(handle);
 234                connection.mutex.Release();
 235                request.semaphore.Wait();
 236                connection.mutex.Wait();
 237                delete request.semaphore;
 238                newPosition = request.readSize;
 239                delete request;
 240             }
 241             connection.mutex.Release();
 242             break;
 243          }
 244       }
 245       if(newPosition != position)
 246       {
 247          if(newPosition > position && newPosition < position + bufferCount)
 248          {
 249             memcpy(buffer, buffer + newPosition - position, bufferCount - (newPosition - position));
 250             bufferCount -= (newPosition - position);
 251          }
 252          else
 253             bufferCount = 0;
 254          position = newPosition;
 255       }
 256       return true;
 257    }
 258
 259    uint Tell()
 260    {
 261       return position;
 262    }
 263
 264    bool Eof()
 265    {
 266       return false;
 267    }
 268
 269    uint GetSize()
 270    {
 271       uint size;
 272       connection.mutex.Wait();
 273       {
 274          Request request = connection.AddRequest(4, Request::SizeCallback, this, true);
 275          connection.SendSizePacket(handle);
 276          connection.mutex.Release();
 277          request.semaphore.Wait();
 278          connection.mutex.Wait();
 279          delete request.semaphore;
 280          size = request.readSize;
 281          delete request;
 282       }
 283       connection.mutex.Release();
 284       return size;
 285    }
 286
 287    // Returns the last request needed to fill this needed size
 288    Request FigureOutMissingData(uint * start, uint * size)
 289    {
 290       Request lastRequest = null;
 291       FileServerConnection connection = this.connection;
 292       OldList blocks { };
 293       Request request;
 294       NetDataBlock block { };
 295
 296       blocks.Add(block);
 297
 298       // Initialize block to what we need
 299       block.start = *start;
 300       block.end = *start + *size - 1;
 301
 302       for(request = connection.requests.first; request; request = request.next)
 303       {
 304          if(request.file == this &&
 305             (request.CallBack == Request::ReadResultCallback || request.CallBack == Request::ReadCallback))
 306          {
 307             // If this request overlap with the entire block of data we need
 308             if(*start + *size >= request.readStart &&
 309                *start + *size < request.readStart + request.readSize)
 310             {
 311                NetDataBlock next;
 312                for(block = blocks.first; block; block = next)
 313                {
 314                   next = block.next;
 315                   // If this block overlap with this request
 316                   if(request.readStart                     < block.end &&
 317                      request.readStart + request.readSize >= block.start)
 318                   {
 319                      // We have all this block, get rid of it
 320                      if(request.readStart <= block.start &&
 321                         request.readStart + request.readSize > block.end)
 322                         blocks.Delete(block);
 323                      // We're missing data at the end of this block
 324                      else if(request.readStart <= block.start)
 325                         block.start = request.readStart;
 326                      // We're missing data at the beginning of this block
 327                      else if(request.readStart + request.readSize > block.end)
 328                         block.end = request.readStart + request.readSize-1;
 329                      lastRequest = request;
 330                   }
 331                }
 332             }
 333          }
 334       }
 335
 336       if(blocks.first)
 337       {
 338          // We still need data
 339          *start = ((NetDataBlock)blocks.first).start;
 340          *size = ((NetDataBlock)blocks.last).end - *start + 1;
 341
 342          blocks.Free(null); // eList_Delete(&blocks, NULL); ??
 343       }
 344       else
 345          *size = 0;
 346       return lastRequest;
 347    }
 348
 349    void ReadMoreData(uint start, uint size, bool wait)
 350    {
 351       FileServerConnection connection = this.connection;
 352       Request lastRequestNeeded;
 353
 354       // If it's not already requested, request it
 355       connection.mutex.Wait();
 356       lastRequestNeeded = FigureOutMissingData(&start, &size);
 357       // If we still need to request more
 358       if(size)
 359       {
 360          Request request = connection.AddRequest(4, Request::ReadResultCallback, this, wait);
 361          request.readStart = start;
 362          request.readSize = size;
 363          connection.SendReadPacket(handle, start, size);
 364          connection.mutex.Release();
 365          if(wait)
 366          {
 367             request.semaphore.Wait();
 368             delete request.semaphore;
 369             delete request;
 370          }
 371       }
 372       else
 373       {
 374          if(wait && lastRequestNeeded)
 375          {
 376             if(!lastRequestNeeded.semaphore)
 377                lastRequestNeeded.semaphore = Semaphore { };
 378          }
 379          connection.mutex.Release();
 380          lastRequestNeeded.semaphore.Wait();
 381       }
 382    }
 383
 384    FileServerConnection connection;
 385    int handle;
 386    byte * buffer;
 387    uint bufferSize;
 388    uint bufferCount;
 389    uint position;
 390 }
 391
 392 public FileServerConnection ConnectToFileServer(char * hostName, int port)
 393 {
 394    FileServerConnection connection { };
 395    if(connection)
 396    {
 397       if(connection.Connect(hostName, port))
 398          connection.fileClientThread.Create();
 399       else
 400          delete connection;
 401    }
 402    return connection;
 403 }
 404
 405 public class FileServerConnection : Socket
 406 {
 407    processAlone = true;
 408 public:
 409    NetworkClientFile Open(char * fileName, FileOpenMode mode)
 410    {
 411       NetworkClientFile f { };
 412       Request request;
 413
 414       f.connection = this;
 415       f.position = 0;
 416
 417       mutex.Wait();
 418       request = AddRequest(4, Request::OpenCallback, f, true);
 419       SendOpenPacket(fileName);
 420       mutex.Release();
 421       request.semaphore.Wait();
 422       delete request.semaphore;
 423       delete request;
 424
 425       if(!f.handle)
 426          delete f;
 427       return f;
 428    }
 429
 430 private:
 431    ~FileServerConnection()
 432    {
 433       if(fileClientThread)
 434       {
 435          fileClientThread.terminate = true;
 436          //guiApp.SignalEvent();
 437          fileClientThread.socket.Disconnect(0);
 438          fileClientThread.Wait();
 439          delete fileClientThread;
 440       }
 441    }
 442
 443    uint OnReceive(const byte * buffer, uint count)
 444    {
 445       Request request;
 446       uint bytesProcessed = 0;
 447
 448       mutex.Wait();
 449       while((request = requests.first))
 450       {
 451          uint numBytes = request.numBytes;
 452          if(count - bytesProcessed < numBytes)
 453             break;
 454
 455          requests.Remove(request);
 456          mutex.Release();
 457
 458          request.CallBack(request, buffer + bytesProcessed);
 459          bytesProcessed += numBytes;
 460
 461          mutex.Wait();
 462       }
 463       mutex.Release();
 464       return bytesProcessed;
 465    }
 466
 467    Request AddRequest(uint numBytes, void (* callback)(Request request, const byte * buffer), NetworkClientFile file, bool createSemaphore)
 468    {
 469       Request request { numBytes = numBytes, CallBack = callback, file = file };
 470       if(createSemaphore)
 471          request.semaphore = Semaphore { };
 472       requests.Add(request);
 473
 474       // There is a possibility that OnReceive already got the data for our request
 475       // If it is the case, and it had no other request, it will have returned 0.
 476       // This mean that OnReceive won't be called again unless more data comes in.
 477       // Signaling an event will unblock the thread to make sure OnReceive is called again.
 478       guiApp.SignalEvent();
 479       return request;
 480    }
 481
 482    // Packet Sending Functions
 483    void SendOpenPacket(char * fileName)
 484    {
 485       int len = strlen(fileName);
 486       OpenPacket packet = (OpenPacket) new0 byte[sizeof(class OpenPacket) + len];
 487       PUTXDWORD(packet.type, NetworkClientPacketType::open);
 488       PUTXDWORD(packet.size, sizeof(class OpenPacket) + len);
 489       PUTXDWORD(packet.fileNameLen, len);
 490       strcpy(packet.fileName, fileName);
 491       Send(packet, GETXDWORD(packet.size));
 492       delete packet;
 493    }
 494
 495    void SendReadPacket(int handle, uint start, uint size)
 496    {
 497       ReadPacket packet
 498       {
 499          XDWORD(NetworkClientPacketType::read),
 500          XDWORD(sizeof(class ReadPacket)),
 501          XDWORD(handle),
 502          XDWORD(start),
 503          XDWORD(size)
 504       };
 505       Send(packet, GETXDWORD(packet.size));
 506    }
 507
 508    void SendSizePacket(int handle)
 509    {
 510       SizePacket packet
 511       {
 512          XDWORD(NetworkClientPacketType::getSize),
 513          XDWORD(sizeof(class SizePacket)),
 514          XDWORD(handle)
 515       };
 516       Send(packet, GETXDWORD(packet.size));
 517    }
 518
 519    FileClientThread fileClientThread { socket = this };
 520    Mutex mutex { };
 521    OldList requests;
 522 }
 523
 524 #endif