Changeset 3890 for trunk/sources/HeuristicLab.Problems.ExternalEvaluation/3.3/Drivers

Timestamp:

06/02/10 16:18:39 (14 years ago)

Author:

abeham

Message:

#866

• Fixed EvaluationTCPChannel
  • Message delimiter and message are sent in the same call to socket.Send
• Added EvaluationTCPChannelView

File:

• trunk/sources/HeuristicLab.Problems.ExternalEvaluation/3.3/Drivers/EvaluationTCPChannel.cs (modified) (2 diffs)

trunk/sources/HeuristicLab.Problems.ExternalEvaluation/3.3/Drivers/EvaluationTCPChannel.cs

@@ -32 +32 @@
   [StorableClass]
   public class EvaluationTCPChannel : EvaluationChannel {
+    public const int MAX_VARINT32_SIZE = 5;
+
     [Storable]
     private string ipAddress;
+    public string IpAddress {
+      get { return ipAddress; }
+      set {
+        bool changed = !ipAddress.Equals(value);
+        ipAddress = value;
+        if (changed)
+          OnIpAddressChanged();
+      }
+    }
     [Storable]
     private int port;
+    public int Port {
+      get { return port; }
+      set {
+        bool changed = port != value;
+        port = value;
+        if (changed)
+          OnPortChanged();
+      }
+    }
     private Socket socket;
 
@@ -57 +77 @@
       socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
       socket.Connect(IPAddress.Parse(ipAddress), port);
-      if (socket.Connected)
+      if (socket.Connected) {
         base.Open();
+        OnConnected();
+      }
     }
 
     public override void Send(IMessage message) {
-      int size = message.SerializedSize;
-      socket.Send(System.BitConverter.GetBytes(size));
-      socket.Send(message.ToByteArray());
+      try {
+        byte[] buffer = EncodeDelimited(message);
+        socket.Send(buffer);
+      } catch (SocketException) {
+        Close();
+      } catch (ObjectDisposedException) {
+        socket = null;
+        Close();
+      }
+    }
+
+    private byte[] EncodeDelimited(IMessage message) {
+      int messageSize = message.SerializedSize;
+      int headerSize = GetVarint32EncodedSize(messageSize);
+      byte[] buffer = new byte[headerSize + messageSize];
+      CodedOutputStream tmp = CodedOutputStream.CreateInstance(buffer);
+      tmp.WriteRawVarint32((uint)messageSize);
+      message.WriteTo(tmp);
+      return buffer;
+    }
+
+    private int GetVarint32EncodedSize(int size) {
+      // Varints in Protocol Buffers are encoded using the 7 lower order bits (the MSB indicates continuation (=1) or termination (=0))
+      int sizeByteCount = 1;
+      int limit = 128;
+      while (size > limit) {
+        sizeByteCount++;
+        limit *= 128;
+      }
+      return sizeByteCount;
     }
 
     public override IMessage Receive(IBuilder builder) {
-      byte[] sizeBuffer = new byte[4];
-      socket.Receive(sizeBuffer);
-      int size = System.BitConverter.ToInt32(sizeBuffer, 0);
-      byte[] messageBuffer = new byte[size];
-      int offset = 0;
-      while (offset < size)
-        offset += socket.Receive(messageBuffer, offset, size, SocketFlags.None);
-      return builder.WeakMergeFrom(ByteString.CopyFrom(messageBuffer)).WeakBuild();
+      try {
+        byte[] buffer = GetMessageBuffer();
+        return builder.WeakMergeFrom(ByteString.CopyFrom(buffer)).WeakBuild();
+      } catch (SocketException) {
+        Close();
+      } catch (ObjectDisposedException) {
+        socket = null;
+        Close();
+      }
+      return null;
+    }
+
+    private byte[] GetMessageBuffer() {
+      byte[] buffer;
+      int messageSize = GetReceivedMessageSize(out buffer);
+      int headerSize = GetVarint32EncodedSize(messageSize);
+      int messageBytesInHeader = Math.Min(messageSize, buffer.Length - headerSize);
+      byte[] messageBuffer = new byte[messageSize];
+      Array.Copy(buffer, headerSize, messageBuffer, 0, messageBytesInHeader);
+      ReceiveMessage(messageBuffer, messageBytesInHeader, messageSize - messageBytesInHeader);
+      return messageBuffer;
+    }
+
+    private int GetReceivedMessageSize(out byte[] buffer) {
+      buffer = new byte[MAX_VARINT32_SIZE];
+      socket.Receive(buffer);
+      CodedInputStream tmp = CodedInputStream.CreateInstance(buffer);
+      return (int)tmp.ReadRawVarint32();
+    }
+
+    private void ReceiveMessage(byte[] buffer, int offset, int size) {
+      while (size > 0) {
+        int received = socket.Receive(buffer, offset, size, SocketFlags.None);
+        offset += received;
+        size -= received;
+      }
     }
 
     public override void Close() {
-      socket.Disconnect(false);
-      socket.Close();
+      if (socket != null) {
+        if (socket.Connected)
+          socket.Disconnect(false);
+        socket.Close();
+        socket = null;
+      }
+      bool wasInitialized = IsInitialized;
       base.Close();
+      if (wasInitialized) OnDiconnected();
     }
 
     #endregion
+
+    #region Events
+    public event EventHandler IpAddressChanged;
+    protected void OnIpAddressChanged() {
+      EventHandler handler = IpAddressChanged;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
+    public event EventHandler PortChanged;
+    protected void OnPortChanged() {
+      EventHandler handler = PortChanged;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
+    public event EventHandler Connected;
+    protected void OnConnected() {
+      EventHandler handler = Connected;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
+    public event EventHandler Disconnected;
+    protected void OnDiconnected() {
+      EventHandler handler = Disconnected;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
+    #endregion
   }
 }