Free cookie consent management tool by TermsFeed Policy Generator

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

Last change on this file since 5445 was 5445, checked in by swagner, 13 years ago

Updated year of copyrights (#1406)

File size: 6.4 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    public const int MAX_VARINT32_SIZE = 5;
35
36    [Storable]
37    private string ipAddress;
38    public string IpAddress {
39      get { return ipAddress; }
40      set {
41        bool changed = !ipAddress.Equals(value);
42        ipAddress = value;
43        if (changed)
44          OnIpAddressChanged();
45      }
46    }
47    [Storable]
48    private int port;
49    public int Port {
50      get { return port; }
51      set {
52        bool changed = port != value;
53        port = value;
54        if (changed)
55          OnPortChanged();
56      }
57    }
58    private Socket socket;
59
60    [StorableConstructor]
61    protected EvaluationTCPChannel(bool deserializing) : base(deserializing) { }
62    protected EvaluationTCPChannel(EvaluationTCPChannel original, Cloner cloner)
63      : base(original, cloner) {
64      ipAddress = original.ipAddress;
65      port = original.port;
66    }
67    public override IDeepCloneable Clone(Cloner cloner) {
68      return new EvaluationTCPChannel(this, cloner);
69    }
70
71    public EvaluationTCPChannel() : this(String.Empty, 0) { }
72    public EvaluationTCPChannel(string ip, int port)
73      : base() {
74      this.ipAddress = ip;
75      this.port = port;
76    }
77
78    #region IExternalEvaluationChannel Members
79
80    public override void Open() {
81      socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
82      socket.Connect(IPAddress.Parse(ipAddress), port);
83      if (socket.Connected) {
84        base.Open();
85        OnConnected();
86      }
87    }
88
89    public override void Send(IMessage message) {
90      try {
91        byte[] buffer = EncodeDelimited(message);
92        socket.Send(buffer);
93      }
94      catch (SocketException) {
95        Close();
96        throw;
97      }
98      catch (ObjectDisposedException) {
99        socket = null;
100        Close();
101        throw;
102      }
103    }
104
105    private byte[] EncodeDelimited(IMessage message) {
106      int messageSize = message.SerializedSize;
107      int headerSize = GetVarint32EncodedSize(messageSize);
108      byte[] buffer = new byte[headerSize + messageSize];
109      CodedOutputStream tmp = CodedOutputStream.CreateInstance(buffer);
110      tmp.WriteRawVarint32((uint)messageSize);
111      message.WriteTo(tmp);
112      return buffer;
113    }
114
115    private int GetVarint32EncodedSize(int size) {
116      // Varints in Protocol Buffers are encoded using the 7 lower order bits (the MSB indicates continuation (=1) or termination (=0))
117      int sizeByteCount = 1;
118      int limit = 128;
119      while (size >= limit) {
120        sizeByteCount++;
121        limit *= 128;
122      }
123      return sizeByteCount;
124    }
125
126    public override IMessage Receive(IBuilder builder) {
127      try {
128        byte[] buffer = GetMessageBuffer();
129        return builder.WeakMergeFrom(ByteString.CopyFrom(buffer)).WeakBuild();
130      }
131      catch (SocketException) {
132        Close();
133        throw;
134      }
135      catch (ObjectDisposedException) {
136        socket = null;
137        Close();
138        throw;
139      }
140    }
141
142    private byte[] GetMessageBuffer() {
143      byte[] buffer;
144      int messageSize = GetReceivedMessageSize(out buffer);
145      int headerSize = GetVarint32EncodedSize(messageSize);
146      int messageBytesInHeader = Math.Min(messageSize, buffer.Length - headerSize);
147      byte[] messageBuffer = new byte[messageSize];
148      Array.Copy(buffer, headerSize, messageBuffer, 0, messageBytesInHeader);
149      ReceiveMessage(messageBuffer, messageBytesInHeader, messageSize - messageBytesInHeader);
150      return messageBuffer;
151    }
152
153    private int GetReceivedMessageSize(out byte[] buffer) {
154      buffer = new byte[MAX_VARINT32_SIZE];
155      socket.Receive(buffer);
156      CodedInputStream tmp = CodedInputStream.CreateInstance(buffer);
157      return (int)tmp.ReadRawVarint32();
158    }
159
160    private void ReceiveMessage(byte[] buffer, int offset, int size) {
161      while (size > 0) {
162        int received = socket.Receive(buffer, offset, size, SocketFlags.None);
163        offset += received;
164        size -= received;
165      }
166    }
167
168    public override void Close() {
169      if (socket != null) {
170        try {
171          if (socket.Connected)
172            socket.Disconnect(false);
173          socket.Close();
174        }
175        catch { }
176        socket = null;
177      }
178      bool wasInitialized = IsInitialized;
179      base.Close();
180      if (wasInitialized) OnDiconnected();
181    }
182
183    #endregion
184
185    #region Events
186    public event EventHandler IpAddressChanged;
187    protected void OnIpAddressChanged() {
188      EventHandler handler = IpAddressChanged;
189      if (handler != null) handler(this, EventArgs.Empty);
190    }
191    public event EventHandler PortChanged;
192    protected void OnPortChanged() {
193      EventHandler handler = PortChanged;
194      if (handler != null) handler(this, EventArgs.Empty);
195    }
196    public event EventHandler Connected;
197    protected void OnConnected() {
198      EventHandler handler = Connected;
199      if (handler != null) handler(this, EventArgs.Empty);
200    }
201    public event EventHandler Disconnected;
202    protected void OnDiconnected() {
203      EventHandler handler = Disconnected;
204      if (handler != null) handler(this, EventArgs.Empty);
205    }
206    #endregion
207  }
208}
Note: See TracBrowser for help on using the repository browser.