Free cookie consent management tool by TermsFeed Policy Generator

source: branches/Async/HeuristicLab.Problems.ExternalEvaluation/3.4/Drivers/EvaluationTCPChannel.cs @ 14534

Last change on this file since 14534 was 12012, checked in by ascheibe, 10 years ago

#2212 merged r12008, r12009, r12010 back into trunk

File size: 6.9 KB
Line 
1#region License Information
2/* HeuristicLab
3 * Copyright (C) 2002-2015 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
4 *
5 * This file is part of HeuristicLab.
6 *
7 * HeuristicLab is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * HeuristicLab is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
19 */
20#endregion
21
22using System;
23using System.Net;
24using System.Net.Sockets;
25using Google.ProtocolBuffers;
26using HeuristicLab.Common;
27using HeuristicLab.Core;
28using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;
29
30namespace HeuristicLab.Problems.ExternalEvaluation {
31  [Item("EvaluationTCPChannel", "A channel that creates a TCP connection over a network.")]
32  [StorableClass]
33  public class EvaluationTCPChannel : EvaluationChannel {
34
35    public const int MAX_VARINT32_SIZE = 5;
36
37    #region Fields & Properties
38    [Storable]
39    private string ipAddress;
40    public string IpAddress {
41      get { return ipAddress; }
42      set {
43        if (value == ipAddress) return;
44        ipAddress = value;
45        UpdateName();
46        OnIpAddressChanged();
47      }
48    }
49    [Storable]
50    private int port;
51    public int Port {
52      get { return port; }
53      set {
54        if (value == port) return;
55        port = value;
56        UpdateName();
57        OnPortChanged();
58      }
59    }
60    private Socket socket;
61    #endregion
62
63    #region Construction & Cloning
64    [StorableConstructor]
65    protected EvaluationTCPChannel(bool deserializing) : base(deserializing) { }
66    protected EvaluationTCPChannel(EvaluationTCPChannel original, Cloner cloner)
67      : base(original, cloner) {
68      ipAddress = original.ipAddress;
69      port = original.port;
70      UpdateName();
71    }
72
73    public override IDeepCloneable Clone(Cloner cloner) {
74      return new EvaluationTCPChannel(this, cloner);
75    }
76
77    public EvaluationTCPChannel() : this(String.Empty, 0) { }
78    public EvaluationTCPChannel(string ip, int port)
79      : base() {
80      this.ipAddress = ip;
81      this.port = port;
82      UpdateName();
83    }
84    [StorableHook(HookType.AfterDeserialization)]
85    private void AfterDeserialization() {
86      UpdateName();
87    }
88    #endregion
89
90
91
92    #region IExternalEvaluationChannel Members
93
94    public override void Open() {
95      socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
96      socket.Connect(IPAddress.Parse(ipAddress), port);
97      if (socket.Connected) {
98        base.Open();
99        OnConnected();
100      }
101    }
102
103    public override void Send(IMessage message) {
104      try {
105        byte[] buffer = EncodeDelimited(message);
106        socket.Send(buffer);
107      } catch (SocketException) {
108        Close();
109        throw;
110      } catch (ObjectDisposedException) {
111        socket = null;
112        Close();
113        throw;
114      }
115    }
116
117    private byte[] EncodeDelimited(IMessage message) {
118      int messageSize = message.SerializedSize;
119      int headerSize = GetVarint32EncodedSize(messageSize);
120      byte[] buffer = new byte[headerSize + messageSize];
121      CodedOutputStream tmp = CodedOutputStream.CreateInstance(buffer);
122      tmp.WriteRawVarint32((uint)messageSize);
123      message.WriteTo(tmp);
124      return buffer;
125    }
126
127    private int GetVarint32EncodedSize(int size) {
128      // Varints in Protocol Buffers are encoded using the 7 lower order bits (the MSB indicates continuation (=1) or termination (=0))
129      int sizeByteCount = 1;
130      int limit = 128;
131      while (size >= limit) {
132        sizeByteCount++;
133        limit *= 128;
134      }
135      return sizeByteCount;
136    }
137
138    public override IMessage Receive(IBuilder builder, ExtensionRegistry extensions) {
139      try {
140        byte[] buffer = GetMessageBuffer();
141        return builder.WeakMergeFrom(ByteString.CopyFrom(buffer), extensions).WeakBuild();
142      } catch (SocketException) {
143        Close();
144        throw;
145      } catch (ObjectDisposedException) {
146        socket = null;
147        Close();
148        throw;
149      }
150    }
151
152    private byte[] GetMessageBuffer() {
153      byte[] buffer;
154      int messageSize = GetReceivedMessageSize(out buffer);
155      int headerSize = GetVarint32EncodedSize(messageSize);
156      int messageBytesInHeader = Math.Min(messageSize, buffer.Length - headerSize);
157      byte[] messageBuffer = new byte[messageSize];
158      Array.Copy(buffer, headerSize, messageBuffer, 0, messageBytesInHeader);
159      ReceiveMessage(messageBuffer, messageBytesInHeader, messageSize - messageBytesInHeader);
160      return messageBuffer;
161    }
162
163    private int GetReceivedMessageSize(out byte[] buffer) {
164      buffer = new byte[MAX_VARINT32_SIZE];
165      socket.Receive(buffer);
166      CodedInputStream tmp = CodedInputStream.CreateInstance(buffer);
167      return (int)tmp.ReadRawVarint32();
168    }
169
170    private void ReceiveMessage(byte[] buffer, int offset, int size) {
171      while (size > 0) {
172        int received = socket.Receive(buffer, offset, size, SocketFlags.None);
173        offset += received;
174        size -= received;
175      }
176    }
177
178    public override void Close() {
179      if (socket != null) {
180        try {
181          if (socket.Connected)
182            socket.Disconnect(false);
183          socket.Close();
184        } catch { }
185        socket = null;
186      }
187      bool wasInitialized = IsInitialized;
188      base.Close();
189      if (wasInitialized) OnDiconnected();
190    }
191
192    #endregion
193
194    #region Auxiliary Methods
195    private void UpdateName() {
196      name = string.Format("TCPChannel {0}:{1}", ipAddress, port);
197      OnNameChanged();
198    }
199    #endregion
200
201    #region Events
202    public event EventHandler IpAddressChanged;
203    protected void OnIpAddressChanged() {
204      EventHandler handler = IpAddressChanged;
205      if (handler != null) handler(this, EventArgs.Empty);
206    }
207    public event EventHandler PortChanged;
208    protected void OnPortChanged() {
209      EventHandler handler = PortChanged;
210      if (handler != null) handler(this, EventArgs.Empty);
211    }
212    public event EventHandler Connected;
213    protected void OnConnected() {
214      EventHandler handler = Connected;
215      if (handler != null) handler(this, EventArgs.Empty);
216    }
217    public event EventHandler Disconnected;
218    protected void OnDiconnected() {
219      EventHandler handler = Disconnected;
220      if (handler != null) handler(this, EventArgs.Empty);
221    }
222    #endregion
223  }
224}
Note: See TracBrowser for help on using the repository browser.