Free cookie consent management tool by TermsFeed Policy Generator

source: trunk/sources/HeuristicLab.Problems.ExternalEvaluation/3.3/Drivers/EvaluationTCPChannel.cs @ 7225

Last change on this file since 7225 was 6470, checked in by epitzer, 13 years ago

Automatically generate TCPChannel, ProcessChannel and ServiceClient names form parameters (#1526).

File size: 6.9 KB
Line 
1#region License Information
2/* HeuristicLab
3 * Copyright (C) 2002-2011 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      }
108      catch (SocketException) {
109        Close();
110        throw;
111      }
112      catch (ObjectDisposedException) {
113        socket = null;
114        Close();
115        throw;
116      }
117    }
118
119    private byte[] EncodeDelimited(IMessage message) {
120      int messageSize = message.SerializedSize;
121      int headerSize = GetVarint32EncodedSize(messageSize);
122      byte[] buffer = new byte[headerSize + messageSize];
123      CodedOutputStream tmp = CodedOutputStream.CreateInstance(buffer);
124      tmp.WriteRawVarint32((uint)messageSize);
125      message.WriteTo(tmp);
126      return buffer;
127    }
128
129    private int GetVarint32EncodedSize(int size) {
130      // Varints in Protocol Buffers are encoded using the 7 lower order bits (the MSB indicates continuation (=1) or termination (=0))
131      int sizeByteCount = 1;
132      int limit = 128;
133      while (size >= limit) {
134        sizeByteCount++;
135        limit *= 128;
136      }
137      return sizeByteCount;
138    }
139
140    public override IMessage Receive(IBuilder builder) {
141      try {
142        byte[] buffer = GetMessageBuffer();
143        return builder.WeakMergeFrom(ByteString.CopyFrom(buffer)).WeakBuild();
144      }
145      catch (SocketException) {
146        Close();
147        throw;
148      }
149      catch (ObjectDisposedException) {
150        socket = null;
151        Close();
152        throw;
153      }
154    }
155
156    private byte[] GetMessageBuffer() {
157      byte[] buffer;
158      int messageSize = GetReceivedMessageSize(out buffer);
159      int headerSize = GetVarint32EncodedSize(messageSize);
160      int messageBytesInHeader = Math.Min(messageSize, buffer.Length - headerSize);
161      byte[] messageBuffer = new byte[messageSize];
162      Array.Copy(buffer, headerSize, messageBuffer, 0, messageBytesInHeader);
163      ReceiveMessage(messageBuffer, messageBytesInHeader, messageSize - messageBytesInHeader);
164      return messageBuffer;
165    }
166
167    private int GetReceivedMessageSize(out byte[] buffer) {
168      buffer = new byte[MAX_VARINT32_SIZE];
169      socket.Receive(buffer);
170      CodedInputStream tmp = CodedInputStream.CreateInstance(buffer);
171      return (int)tmp.ReadRawVarint32();
172    }
173
174    private void ReceiveMessage(byte[] buffer, int offset, int size) {
175      while (size > 0) {
176        int received = socket.Receive(buffer, offset, size, SocketFlags.None);
177        offset += received;
178        size -= received;
179      }
180    }
181
182    public override void Close() {
183      if (socket != null) {
184        try {
185          if (socket.Connected)
186            socket.Disconnect(false);
187          socket.Close();
188        }
189        catch { }
190        socket = null;
191      }
192      bool wasInitialized = IsInitialized;
193      base.Close();
194      if (wasInitialized) OnDiconnected();
195    }
196
197    #endregion
198
199    #region Auxiliary Methods
200    private void UpdateName() {
201      name = string.Format("TCPChannel {0}:{1}", ipAddress, port);
202      OnNameChanged();
203    }
204    #endregion
205
206    #region Events
207    public event EventHandler IpAddressChanged;
208    protected void OnIpAddressChanged() {
209      EventHandler handler = IpAddressChanged;
210      if (handler != null) handler(this, EventArgs.Empty);
211    }
212    public event EventHandler PortChanged;
213    protected void OnPortChanged() {
214      EventHandler handler = PortChanged;
215      if (handler != null) handler(this, EventArgs.Empty);
216    }
217    public event EventHandler Connected;
218    protected void OnConnected() {
219      EventHandler handler = Connected;
220      if (handler != null) handler(this, EventArgs.Empty);
221    }
222    public event EventHandler Disconnected;
223    protected void OnDiconnected() {
224      EventHandler handler = Disconnected;
225      if (handler != null) handler(this, EventArgs.Empty);
226    }
227    #endregion
228  }
229}
Note: See TracBrowser for help on using the repository browser.