Changeset 9439 for trunk/MagicSoft/Cosy/tcpip/MTcpIpIO.cc

Timestamp:

05/09/09 13:48:12 (16 years ago)

Author:

tbretz

Message:

*** empty log message ***

File:

• trunk/MagicSoft/Cosy/tcpip/MTcpIpIO.cc (modified) (8 diffs)

trunk/MagicSoft/Cosy/tcpip/MTcpIpIO.cc

@@ -43 +43 @@
 MTcpIpO::MTcpIpO(const char *addr, Int_t tx) : fPortTx(tx)
 {
-    gLog << inf2 << "- Open send socket to " << addr << ":" << tx << endl;
+    // was "inf2" but nobody knows what the prg is doing if it times out
+    gLog << all << "- Open send socket to " << addr << ":" << tx << endl;
+
     fTxSocket = new TSocket(addr, tx);
+    fTxSocket->SetOption(kNoBlock, 1);
+
+    MTcpIpO::RunThread();
 }
 
 MTcpIpO::~MTcpIpO()
 {
+    CancelThread();
+
     // Now delete all TCP/IP objects
     delete fTxSocket;
@@ -55 +62 @@
 MTcpIpIO::MTcpIpIO(const char *addr, Int_t tx, Int_t rx) : MTcpIpI(rx), MTcpIpO(addr, tx)
 {
-    RunThread();
+    MTcpIpI::RunThread();
 }
 
@@ -74 +81 @@
 }
 
+/*
 TString MTcpIpO::GetSocketAddress() const
 {
+    TLockGuard guard(const_cast<TMutex*>(&fMutex));
     return GetSocketAddress(*fTxSocket);
 }
-
+*/
 bool MTcpIpO::SendFrame(TSocket &tx, const char *msg, int len)
 {
     if (!tx.IsValid())
-    {
-        //gLog << warn << "WARNING - No transmission to " << GetSocketAddress(tx) << " possible." << endl;
         return false;
-    }
 
 #ifdef DEBUG
@@ -91 +97 @@
 #endif
 
-    const Int_t l = tx.SendRaw(msg, len);
+    const Int_t l = tx.SendRaw(msg, len, kDontBlock);
+
+    // Frame not sent, EWOULDBLOCK
+    if (l==-4)
+    {
+        gLog << err << "ERROR - Sending to " << GetSocketAddress(tx) << " would block." << endl;
+/*--->*/        tx.Close(); // ?????
+        return false;
+    }
+
     if (l<0)
     {
@@ -100 +115 @@
     if (l!=len)
     {
-        gLog << err << "ERROR - Sent wrong number (" << l << ") of bytes to " << GetSocketAddress(tx) << endl;
+        gLog << err << "ERROR - Could only sent " << l << " out of "  << len << " bytes to " << GetSocketAddress(tx) << endl;
         return false;
     }
@@ -109 +124 @@
 bool MTcpIpO::SendFrame(const char *addr, int port, const char *msg, int len)
 {
-    //    R__LOCKGUARD2(myMutex);
 #ifdef DEBUG
     cout << "Connecting to " << addr << ":" << port << endl;
@@ -117 +131 @@
     TSocket tx(addr, port);
     return SendFrame(tx, msg, len);
-/*
-    if (!tx.IsValid())
-    {
-        gLog << warn << "WARNING - No transmission to " << addr << ":" << port << " possible." << endl;
+}
+
+bool MTcpIpO::Send(const char *msg, Int_t len)
+{
+    const Int_t mtx = fMutex.TryLock();
+    if (mtx==13)
+        gLog << warn << "MTcpIpO::Send - mutex is already locked by this thread." << endl;
+
+    // If Mutex cannot be locked, i.e. we are currently reopening
+    // the send socket we cannot wait, because we would block the
+    // executing threrad.
+    if (mtx)
         return false;
-    }
-
-    gLog << dbg << "Sending to " << addr << ":" << port << endl;
-
-    const Int_t l = tx.SendRaw(msg, len, kDontBlock);
-    if (l<0)
-    {
-        gLog << err << "ERROR - Sending TCP/IP frame to " << addr << ":" << port << endl;
-        return false;
-    }
-
-    if (l!=len)
-    {
-        gLog << err << "ERROR - Sent wrong number (" << l << ") of bytes to " << addr << ":" << port << endl;
-        return false;
-    }
-
-#ifdef DEBUG
-    cout << "Tx: " << msg << flush;
-#endif
-
-    return true;
-    */
-}
-
-bool MTcpIpO::Send(const char *msg, Int_t len)
-{
-    if (!fTxSocket->IsValid())
-    {
-        const TInetAddress &a = fTxSocket->GetInetAddress();
-        if (!a.IsValid())
-            return false;
+
+    const bool rc = SendFrame(*fTxSocket, msg, len);
+
+    fMutex.UnLock();
+
+    return rc;
+}
+
+Int_t MTcpIpO::Thread()
+{
+    const TInetAddress &a = fTxSocket->GetInetAddress();
+    if (!a.IsValid())
+    {
+        gLog << err << "ERROR: Send socket address invalid." << endl;
+        return 0;
+    }
+
+    Int_t wait = 1000; /// Wait a ms
+
+    while (!IsThreadCanceled())
+    {
+        if (fTxSocket->IsValid())
+        {
+            MThread::Sleep(1000); // Wait a ms
+            wait = 1000;
+            continue;
+        }
+
+
 #ifdef DEBUG
         cout << "- Reopen send socket to " << a.GetHostAddress() << ":" << fPortTx << endl;
 #endif
+
+        fMutex.Lock();
+
         delete fTxSocket;
+
         fTxSocket = new TSocket(a.GetHostAddress(), fPortTx);
-    }
-
-    return SendFrame(*fTxSocket, msg, len);
+        fTxSocket->SetOption(kNoBlock, 1);
+
+        fMutex.UnLock();
+
+        if (fTxSocket->IsValid())
+            continue;
+
+        MThread::Sleep(wait); // Wait a ms
+        if (wait<30000000) // 30s
+            wait *= 2;
+    }
+
+    return 1;
 }
 
@@ -174 +207 @@
     TString str;
 
-    while (!IsThreadCanceled())
+    while (!MTcpIpI::IsThreadCanceled())
     {
         char c;