From 8d10e8bbd1e19adc7c70e1101dbb69c213c910dd Mon Sep 17 00:00:00 2001
From: fcolin
Date: Fri, 22 Aug 2008 16:44:01 +0000
Subject: optimisation for parsing same regular expression from multiple client
using fxCop for code beauty fix bug on concurrent connect
---
Ivy/IvyClient.cs | 373 ++++++++++++++++++++++++++++++-------------------------
1 file changed, 202 insertions(+), 171 deletions(-)
(limited to 'Ivy/IvyClient.cs')
diff --git a/Ivy/IvyClient.cs b/Ivy/IvyClient.cs
index 63dc77e..7f29e91 100644
--- a/Ivy/IvyClient.cs
+++ b/Ivy/IvyClient.cs
@@ -16,6 +16,8 @@ namespace IvyBus
using System.Net.Sockets;
using System.Configuration;
using System.Diagnostics;
+ using System.Collections.ObjectModel;
+ using IvyBus.Properties;
/// A Class for the the peers on the bus.
///
@@ -29,7 +31,10 @@ namespace IvyBus
{
public int CompareTo(IvyClient other)
{
- return (other.clientPriority - clientPriority);
+ if (other == null)
+ return clientPriority;
+ else
+ return (other.clientPriority - clientPriority);
}
public String ApplicationName
@@ -41,20 +46,20 @@ namespace IvyBus
}
- public List Regexps
- {
- get
- {
- List tab = new List();
- lock (bindings)
- {
- foreach (IvyBindingBase bind in bindings.Values)
- tab.Add(bind.Expression);
- }
- return tab;
- }
+ //public ReadOnlyCollection Regexps
+ //{
+ // get
+ // {
+ // List tab = new List();
+ // lock (bindings)
+ // {
+ // foreach (IvyBindingBase bind in bindings.Values)
+ // tab.Add(bind.Expression);
+ // }
+ // return new ReadOnlyCollection(tab);
+ // }
- }
+ //}
internal int AppPort
{
get
@@ -81,16 +86,18 @@ namespace IvyBus
}
private Ivy bus;
- private Dictionary bindings;
+ //private Dictionary bindings;
private int appPort;
- private string clientId; /* an unique ID for each IvyClient */
+ private string clientId; /* an unique ID for each IvyClient */
private int clientPriority; /* client priority */
-
+
private volatile Thread clientThread; // volatile to ensure the quick communication
private bool doping; // false by runtime default
private const int PINGTIMEOUT = 5000;
private volatile Thread pingerThread;
+ private int localPort;
+
private int remotePort;
private IPAddress remoteHost;
@@ -98,38 +105,31 @@ namespace IvyBus
internal String appName;
internal IvyProtocol stream;
- internal IvyClient(Ivy bus, Socket socket, string appname)
+ internal IvyClient(Ivy bus, Socket socket, string appname, int appPort)
{
- bindings = new Dictionary();
- appName = appname;
+ //bindings = new Dictionary();
+ this.appName = appname;
+ this.appPort = appPort;
this.bus = bus;
+
+ localPort = ((IPEndPoint)socket.LocalEndPoint).Port;
IPEndPoint endpoint = (IPEndPoint)socket.RemoteEndPoint;
-
remoteHost = endpoint.Address;
remotePort = endpoint.Port;
-#if (!PocketPC )
- // try to position Keep Alive
- //try
- //{
- // socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.KeepAlive, 1);
- //}
- //catch (SocketException)
- //{
- //}
-#endif
-
if ( bus.ProtocolVersion == 4 )
stream = new IvyTCPStreamV4( socket, this );
else
stream = new IvyTCPStreamV3(socket, this);
- clientPriority = Ivy.DEFAULT_PRIORITY;
- // spawns a thread to manage the incoming traffic on this
+ // spawns a thread to manage the incoming traffic on this
// socket. We should be ready to receive messages now.
clientThread = new Thread(new ThreadStart(this.Run));
clientThread.Name = "Ivy Tcp Client Reader Thread ("+appname+")";
+
+ bus.addClient(this);
+
clientThread.Start();
}
@@ -137,44 +137,62 @@ namespace IvyBus
internal void SendBindings()
{
try
- {
- stream.TokenApplicationId(bus.applicationPriority, bus.AppId);
-
- // sends our ID, whether we initiated the connexion or not
- // the ID is the couple "host name,application Port", the host name
- // information is in the socket itself, the port is not known if we
- // initiate the connexion
- stream.TokenStartRegexp(bus.applicationPort, bus.appName);
- // sends our regexps to the peer
- lock (bus.bindings)
{
- foreach (IvyApplicationBinding bind in bus.bindings.Values)
+
+ // sends our ID, whether we initiated the connexion or not
+ // the ID is the couple "host name,application Port", the host name
+ // information is in the socket itself, the port is not known if we
+ // initiate the connexion
+ stream.TokenStartRegexp(bus.applicationPort, bus.appName);
+ // sends our regexps to the peer
+ lock (bus.bindings)
{
- stream.TokenAddBinding(bind.Binding, bind.Key, bind.FormatedExpression);
+ foreach (IvyApplicationBinding bind in bus.bindings.Values)
+ {
+ stream.TokenAddBinding(bind.Binding, bind.Key, bind.FormattedExpression);
+ }
}
- }
- stream.TokenEndRegexp();
-
+ // send end of bindings peers can now send ReadyMessage
+ stream.TokenEndRegexp();
+
#if (!PocketPC)
- doping = Properties.Settings.Default.IvyPing;
+ doping = Properties.Settings.Default.IvyPing;
#endif
- if (doping)
- {
- pingerThread = new Thread(new ThreadStart(PingerRun));
- pingerThread.Name = "Ivy Pinger Thread";
- pingerThread.Start();
- }
+ if (doping)
+ {
+ pingerThread = new Thread(new ThreadStart(PingerRun));
+ pingerThread.Name = "Ivy Pinger Thread";
+ pingerThread.Start();
+ }
- }
- catch (IOException ex)
- { // the client nous a coupé l'herbe sous le pied
- Ivy.traceError("IvyClient","I can't send my message to this client. He probably left " + ex.Message);
- // invokes the Disconnected applicationListeners
- //bus.OnClientDisconnected(new IvyEventArgs(this,id, message ));
- // called by the receiver Thread
- close(false);
- }
+ }
+ catch (NullReferenceException ex)
+ {
+ // the client nous a coupé l'herbe sous le pied
+ Ivy.traceError(Resources.IvyClient, Resources.IvyClientLeft + " " + appName + " " + ex.Message);
+ // connexion close in rare concurrent connexion
+ close(false);
+ }
+ catch (ObjectDisposedException ex)
+ {
+ // the client nous a coupé l'herbe sous le pied
+ Ivy.traceError(Resources.IvyClient, Resources.IvyClientLeft + " " + appName + " " + ex.Message);
+ // invokes the Disconnected applicationListeners
+ //bus.OnClientDisconnected(new IvyEventArgs(this,id, message ));
+ // called by the receiver Thread
+ close(false);
+ }
+ catch (IOException ex)
+ {
+ // the client nous a coupé l'herbe sous le pied
+ Ivy.traceError(Resources.IvyClient, Resources.IvyClientLeft + " " + appName + " " + ex.Message);
+ // invokes the Disconnected applicationListeners
+ //bus.OnClientDisconnected(new IvyEventArgs(this,id, message ));
+ // called by the receiver Thread
+ close(false);
+ }
+
}
/// returns the name of the remote agent.
@@ -191,7 +209,7 @@ namespace IvyBus
/// the string that will be match-tested
///
///
- public void SendDirectMsg(ushort id, string message)
+ public void SendDirectMsg(int id, string message)
{
try
{
@@ -199,7 +217,7 @@ namespace IvyBus
}
catch (IOException ex)
{
- Ivy.traceError("IvyClient","I can't send my message to this client. He probably left "+ex.Message);
+ Ivy.traceError(Resources.IvyClient,Resources.IvyClientLeft+ex.Message);
// first, I'm not a first class IvyClient any more
bus.removeClient(this);
// invokes the Disconnected applicationListeners
@@ -217,19 +235,27 @@ namespace IvyBus
///
internal void close(bool notify)
{
- Ivy.traceProtocol("IvyClient","closing connexion to " + appName);
- if (doping )
+ Ivy.traceProtocol(Resources.IvyClient,Resources.Closing + appName);
+ if (doping )
{
StopPinging();
}
if (notify)
try
{
- stream.TokenBye(0, "hasta la vista");
+ if (stream != null)
+ stream.TokenBye(0, Resources.ByeMessage);
+ }
+ catch (ObjectDisposedException)
+ {
}
catch (IOException ioe)
{
- throw new IvyException(ioe.Message);
+ if (!(ioe.InnerException is SocketException))
+ throw new IvyException(ioe.Message);
+ if (((SocketException)ioe.InnerException).SocketErrorCode != SocketError.ConnectionReset)
+ throw new IvyException(ioe.Message);
+
}
// stop the thread and close the stream
if (clientThread == null)
@@ -268,36 +294,37 @@ namespace IvyBus
/// the number of messages sent to the peer
///
///
- internal int sendMsg(String message)
+ internal int sendMsg(ushort id, string[] args)
{
int count = 0;
-
- lock( bindings )
+
+
+ try
{
- try
- {
- foreach (IvyBindingBase bind in bindings.Values)
- {
- string[] args = bind.Match(message);
- if (stream != null && args != null)
- {
- stream.TokenMsg(bind.Key, args);
- count++;
- }
- }
- }
- catch (IOException ex)
- {
- Ivy.traceError("IvyClient","I can't send my message to this client. He probably left " + ex.Message);
- // first, I'm not a first class IvyClient any more
- bus.removeClient(this);
- // invokes the Disconnected applicationListeners
- // in the receiver thread
- close(false);
- }
+ stream.TokenMsg(id, args);
+ count++;
+ }
+ catch (ObjectDisposedException ex)
+ {
+ Ivy.traceError(Resources.IvyClient, Resources.IvyClientLeft + ex.Message);
+ // first, I'm not a first class IvyClient any more
+ bus.removeClient(this); //TODO trouble in upper loop iter
+ // invokes the Disconnected applicationListeners
+ // in the receiver thread
+ close(false);
+ }
+ catch (IOException ex)
+ {
+ Ivy.traceError(Resources.IvyClient, Resources.IvyClientLeft + ex.Message);
+ // first, I'm not a first class IvyClient any more
+ bus.removeClient(this); //TODO trouble in upper loop iter
+ // invokes the Disconnected applicationListeners
+ // in the receiver thread
+ close(false);
+ }
- }
+
return count;
}
@@ -309,25 +336,28 @@ namespace IvyBus
/// © ® (tm)
///
///
- internal bool sameClient(IvyClient clnt)
+ internal bool sameIvyClient(IvyClient clnt)
{
- return (appPort != 0 && appPort == clnt.appPort) && (RemoteAddress == clnt.RemoteAddress);
+ // clientId est null si le protocol n'envoie pas le client ID
+ if (clnt.clientId != null && this.clientId != null && clnt.clientId == this.clientId)
+ return true;
+ return (this.appPort != 0) && (this.appPort == clnt.appPort) && (this.RemoteAddress.Equals(clnt.RemoteAddress));
}
/// the code of the thread handling the incoming messages.
///
private void Run()
{
- Ivy.traceProtocol("IvyClient","Connected from " + RemoteAddress + ":" + RemotePort);
+ Ivy.traceProtocol(Resources.IvyClient, Resources.Connected + RemoteAddress + ":" + RemotePort);
- Ivy.traceProtocol("IvyClient","Thread started");
-
- bool running = true;
+ Ivy.traceProtocol(Resources.IvyClient,"Thread started");
+
+ bool running = true;
while ( running && (stream != null) )
{
try
{
- if ( stream.receiveMsg() )
+ if ( stream.ReceiveMsg() )
{
// early stop during readLine()
if (doping && (pingerThread != null))
@@ -335,26 +365,26 @@ namespace IvyBus
}
else
{
- Ivy.traceProtocol("IvyClient","receiveMsg false ! leaving the thread");
+ Ivy.traceProtocol(Resources.IvyClient, Resources.BadReceive);
running = false;
break;
}
}
catch ( ObjectDisposedException ex )
{
- Ivy.traceError("IvyClient", "socket closed "+ex.Message );
+ Ivy.traceError(Resources.IvyClient, Resources.SocketClosed + ex.Message);
running = false;
break;
}
catch (IvyException ie)
{
- Ivy.traceError("IvyClient","socket closed IvyException" + ie.Message);
+ Ivy.traceError(Resources.IvyClient, Resources.SocketClosed + ie.Message);
running = false;
break;
}
catch (SocketException se)
{
- Ivy.traceError("IvyClient", "socket closed "+se.Message );
+ Ivy.traceError(Resources.IvyClient, Resources.SocketClosed + se.Message );
running = false;
break;
}
@@ -362,21 +392,21 @@ namespace IvyBus
{
if ( ex.InnerException is SocketException )
{
- Ivy.traceProtocol("IvyClient", "socket closed" );
+ Ivy.traceProtocol(Resources.IvyClient, Resources.SocketClosed );
}
else
{
- Ivy.traceError("IvyClient","abnormally Disconnected from " + RemoteAddress + ":" + RemotePort);
+ Ivy.traceError(Resources.IvyClient, Resources.AbDisconnect + RemoteAddress + ":" + RemotePort);
}
running = false;
break;
}
}
- Ivy.traceProtocol("IvyClient","normally Disconnected from " + appName);
- Ivy.traceProtocol("IvyClient","Thread stopped");
+ Ivy.traceProtocol(Resources.IvyClient,Resources.Disconnected + appName);
+ Ivy.traceProtocol(Resources.IvyClient,"Thread stopped");
// invokes the Disconnected applicationListeners
- bus.OnClientDisconnected(new IvyEventArgs(this,0, "" ));
+ bus.OnClientDisconnected(new IvyEventArgs(this,0, string.Empty ));
// first, I'm not a first class IvyClient any more
if (stream != null)
{
@@ -390,14 +420,14 @@ namespace IvyBus
{
// never call in this side
}
- bool IvyProtocol.receiveMsg()
+ bool IvyProtocol.ReceiveMsg()
{
// nerver call in this side
return false;
}
- void IvyProtocol.TokenDie(ushort id, string arg)
+ void IvyProtocol.TokenDie(int id, string arg)
{
- Ivy.traceProtocol("IvyClient","received die Message from " + appName + "Raison: "+ arg);
+ Ivy.traceProtocol(Resources.IvyClient, Resources.DieReceive + appName + Resources.Reason + arg);
// invokes the die applicationListeners
IvyDieEventArgs ev = new IvyDieEventArgs(this, id, arg);
bus.OnDie(ev);
@@ -413,10 +443,10 @@ namespace IvyBus
System.Environment.Exit(0);
#endif
}
- void IvyProtocol.TokenBye(ushort err, string arg)
+ void IvyProtocol.TokenBye(int err, string arg)
{
// the peer quits
- Ivy.traceProtocol("IvyClient","received bye Message from " + appName + "Raison: "+ arg);
+ Ivy.traceProtocol(Resources.IvyClient, Resources.ByeReceive + appName + Resources.Reason + arg);
// first, I'm not a first class IvyClient any more
//bus.removeClient(this); // this is done in the receive thread terminaison!!
// invokes the disconnect applicationListeners
@@ -424,10 +454,10 @@ namespace IvyBus
close(false); // will fire disconnected
}
- void IvyProtocol.TokenAddBinding(BindingType type, ushort id, string expression)
+ void IvyProtocol.TokenAddBinding(BindingType type, int id, string expression)
{
- if (type == BindingType.Regexp && !bus.CheckRegexp(expression))
+ if (type == BindingType.RegularExpression && !bus.CheckRegexp(expression))
{
bus.OnClientFilterBinding(new IvyEventArgs(this, id, expression ));
return;
@@ -437,20 +467,14 @@ namespace IvyBus
{
switch (type)
{
- case BindingType.Regexp:
- bind = new IvyBindingRegexp(id, expression);
+ case BindingType.RegularExpression:
+ bind = new IvyBindingRegexp(expression);
break;
case BindingType.Simple:
- bind = new IvyBindingSimple(id, expression);
+ bind = new IvyBindingSimple(expression);
break;
}
- lock (bindings)
- {
- bindings.Add(id, bind);
- }
-
- bus.OnClientAddBinding(new IvyEventArgs(this, id, expression));
-
+ bus.AddBinding(id, this, bind);
}
catch (ArgumentException ex)
{
@@ -458,81 +482,88 @@ namespace IvyBus
}
}
- void IvyProtocol.TokenDelBinding(ushort id)
+
+
+ void IvyProtocol.TokenDelBinding(int id)
{
- lock( bindings )
- {
- try
- {
- IvyBindingBase bind = bindings[id];
- bus.OnClientRemoveBinding(new IvyEventArgs(this, bind.Key, bind.Expression));
- bindings.Remove(id);
- }
- catch (KeyNotFoundException ex)
- {
- Ivy.traceError("IvyClient","DelBinding " + ex.Message);
- }
- }
+ bus.DelBinding(id, this);
}
- void IvyProtocol.TokenMsg(ushort id, string[] args)
+ void IvyProtocol.TokenMsg(int id, string[] args)
{
bus.OnMessage(new IvyMessageEventArgs(this, id, args));
}
- void IvyProtocol.TokenError(ushort id, string arg)
+ void IvyProtocol.TokenError(int id, string arg)
{
bus.OnError(new IvyEventArgs(this, id, arg));
- Ivy.traceError("IvyClient","Error msg " + id + " " + arg);
- }
- void IvyProtocol.TokenApplicationId(ushort id, string arg)
- {
- clientId = arg;
- if ( clientPriority != id )
- {
- clientPriority = id;
- bus.SortClients();
- }
+ Ivy.traceError(Resources.IvyClient,Resources.ErrorReceive + id + " " + arg);
}
+
void IvyProtocol.TokenEndRegexp()
{
+ bus.OnClientConnected(new IvyEventArgs(this, 0, string.Empty));
/*
* the peer is perhaps not ready to handle this message
* an assymetric processing should be written
*/
if (bus.ReadyMessage != null)
- sendMsg(bus.ReadyMessage);
- bus.OnClientConnected(new IvyEventArgs(this, 0, ""));
+ {
+ bus.SendMsgToClient(this, bus.ReadyMessage);
+ }
+
}
- void IvyProtocol.TokenStartRegexp(ushort id, string arg)
+ void IvyProtocol.TokenStartRegexp(int id, string arg)
{
+ //bool bindingToSend = appPort == 0;
appName = arg;
appPort = id;
- if (bus.checkConnected(this))
- {
- close(false);
- throw new IvyException("Rare ! A concurrent connect occured");
- }
+ IvyClient target = this;
+ IvyClient client = bus.checkConnected(this);
+ if (client != null)
+ {
+ // Dilemma choose the rigth client to close
+ // the symetric processing will try to close each other
+ // only one side may be closed
+ //Console.WriteLine(" should close {0} this local {1} rem {2} other local {3} rem {4}", this.appName, this.localPort, this.remotePort, client.localPort, client.remotePort);
+ if (Math.Max(client.localPort, client.remotePort) > Math.Max( this.localPort, this.remotePort ))
+ {
+ target = client;
+ //Console.WriteLine("choose {0} other ports {1},{2}", target.appName, target.localPort, target.remotePort);
+ }
+ else
+ {
+ target = this;
+ //Console.WriteLine("choose {0} this ports {1},{2}", target.appName, target.remotePort, target.localPort);
+ }
+ bus.removeClient(target);
+ target.close(false);
+ //throw new IvyException(Resources.ConcurrentConnect + " " + appName + " " + clientId);
+
+ }
+ //if ( bindingToSend && target != this)
+ // SendBindings();
+
}
- void IvyProtocol.TokenDirectMsg(ushort id, string arg)
+ void IvyProtocol.TokenDirectMsg(int id, string arg)
{
bus.OnDirectMessage(new IvyEventArgs(this,id,arg));
}
void IvyProtocol.TokenPing(string arg)
{
// I receive a ping. I can answer a pong.
- Ivy.traceProtocol("IvyClient","Ping msg from " + appName + " : " + arg );
+ Ivy.traceProtocol(Resources.IvyClient, Resources.PingReceive + appName + " : " + arg );
stream.TokenPong(arg);
}
void IvyProtocol.TokenPong(string arg)
{
- Ivy.traceProtocol("IvyClient","Ping msg from " + appName + " : " + arg);
+ Ivy.traceProtocol(Resources.IvyClient, Resources.PingReceive + appName + " : " + arg);
}
public override String ToString()
{
- return "IvyClient " + bus.appName + ":" + appName;
+ return Resources.IvyClient+ " " + bus.appName + ":" + appName;
}
/* is the Pinging Thread Runninng */
@@ -541,20 +572,20 @@ namespace IvyBus
private void PingerRun()
{
isPinging = true;
- Ivy.traceProtocol("IvyClient","Pinger Thread started");
+ Ivy.traceProtocol(Resources.IvyClient,Resources.PingerThreadStarted);
while (isPinging)
{
try
{
Thread.Sleep(PINGTIMEOUT);
- stream.TokenPing("are you here ?");
+ stream.TokenPing(Resources.PingerThreadMessage);
}
catch (ThreadAbortException ie)
{
- Ivy.traceError("IvyClient","Pinger Thread killed "+ie.Message);
+ Ivy.traceError(Resources.IvyClient,Resources.PingerThreadKilled + ie.Message);
}
}
- Ivy.traceProtocol("IvyClient","Pinger Thread stopped");
+ Ivy.traceProtocol(Resources.IvyClient,Resources.PingerThreadStopped);
}
public virtual void StopPinging()
{
--
cgit v1.1