CVS update of carob (2 files)

    Date: Thursday, February 2, 2006 @ 18:19:18
  Author: gilles
    Path: /cvsroot/carob/carob

Modified: include/ControllerConnectPolicy.hpp (1.4 -> 1.5)
          src/ControllerConnectPolicy.cpp (1.6 -> 1.7)

Implemented updateSuspectList()
Suspect list is now a vector of struct{controllerInfo/javaSocket*} (was a map 
but appeared to be unusefull)
Fixed getController() from round robin policy: we cannot do like in java 
because our suspect list is shared !
JavaSocket is now a pointer in the suspect list (so we don't "copy" socket 
anymore !)
Added select timeout customization parameter to updateSuspectList()
Added comment on ignored retryIntervalInMs param + set default


-------------------------------------+
 include/ControllerConnectPolicy.hpp |   19 ++-
 src/ControllerConnectPolicy.cpp     |  184 ++++++++++++++++++++++++++--------
 2 files changed, 154 insertions(+), 49 deletions(-)


Index: carob/include/ControllerConnectPolicy.hpp
diff -u carob/include/ControllerConnectPolicy.hpp:1.4 
carob/include/ControllerConnectPolicy.hpp:1.5
--- carob/include/ControllerConnectPolicy.hpp:1.4       Wed Jan 25 23:13:30 2006
+++ carob/include/ControllerConnectPolicy.hpp   Thu Feb  2 18:19:18 2006
@@ -21,6 +21,8 @@
 #ifndef CONTROLLERCONNECTPOLICY_H_
 #define CONTROLLERCONNECTPOLICY_H_
 
+#include "ConnectionParameters.hpp" // for ControllerInfo vector
+
 #include "CarobException.hpp"
 #include "CriticalSection.hpp"
 
@@ -44,11 +46,11 @@
    * Creates a new <code>AbstractControllerConnectPolicy</code> object
    * 
    * @param controllerList the controller list on which the policy applies
-   * @param retryIntervalInMs Interval in milliseconds before retrying to
-   *          re-connect to a controller that has failed
+   * @param retryIntervalInMs IGNORED ! interval in milliseconds before 
retrying
+   *          to re-connect to a controller that has failed
    */
   AbstractControllerConnectPolicy(const std::vector<ControllerInfo>& 
controllerList,
-      long retryIntervalInMs) throw (DriverException, UnexpectedException);
+      long retryIntervalInMs = 0) throw (DriverException, UnexpectedException);
   /**
    * Kills the controller ping thread and clear lists
    */
@@ -78,8 +80,10 @@
    * Tries to connect to each suspected controller with a non-blocking connect
    * and checks the connection state. In case of connection success, send the
    * ping command and remove the controller from the suspect list.
+   * @param select_timeout number of milliseconds to wait for the controllers 
to
+   *        respond to a ping connection attempt (default is 20ms)
    */
-  static void                 updateSuspectList();
+  static void                 updateSuspectList(int select_timeout = 20);
   /**
    * Returns true if the specified controller is suspected of failure.
    * @param controllerInfo the controller to check
@@ -95,10 +99,11 @@
   void                        suspectControllerOfFailure(
                                   ControllerInfo& controllerInfo);
   /**
-   * Removes the specified controller from the list of suspect controllers
+   * Removes the specified controller from the list of suspect controllers.
+   * If the given controller is not in the list, logs a warning and returns
    * @param controller the controller to remove from the list
    */
-  void                        removeControllerFromSuspectList(
+  static void                 removeControllerFromSuspectList(
                                   const ControllerInfo& controller);
   /**
    * Gives the number of controller failures since this process started
@@ -122,7 +127,7 @@
 {
 public:
   RoundRobinConnectPolicy(const std::vector<ControllerInfo>& controllerList,
-      long retryIntervalInMs) throw (DriverException, UnexpectedException);
+    long retryIntervalInMs = 0) throw (DriverException, UnexpectedException);
   ControllerInfo              getController() throw (NoMoreControllerException,
                                   UnexpectedException);
 private:
Index: carob/src/ControllerConnectPolicy.cpp
diff -u carob/src/ControllerConnectPolicy.cpp:1.6 
carob/src/ControllerConnectPolicy.cpp:1.7
--- carob/src/ControllerConnectPolicy.cpp:1.6   Wed Feb  1 16:54:51 2006
+++ carob/src/ControllerConnectPolicy.cpp       Thu Feb  2 18:19:18 2006
@@ -22,11 +22,13 @@
 #include "ControllerConnectPolicy.hpp"
 
 #include "JavaSocket.hpp"
+#include "Connection.hpp"
 
 #include "ConnectionParameters.hpp"
 #include "Common.hpp"
 
 #include <map>
+#include <sstream>
 
 using std::wstring;
 
@@ -34,22 +36,20 @@
 
 CriticalSection AbstractControllerConnectPolicy::suspected_controllers_CS;
 
-// ControllerInfo comparison function
-struct SortByFailureNumber
+typedef struct
 {
-  bool operator()(const ControllerInfo& ci1, const ControllerInfo& ci2) const
-  {
-    return true;//(ci1.getFailureNumber() < ci2.getFailureNumber());
-  }
-};
-
-typedef std::map<ControllerInfo, JavaSocket, SortByFailureNumber> 
SuspectedList;
-// Static list of suspected controllers
+  ControllerInfo  controllerInfo;
+  JavaSocket*     pingSocketPtr;
+} SuspectController;
+
+/** Utility typedef to have a short name for suspect_list content type */
+typedef std::vector<SuspectController> SuspectList;
+/** The list of suspected controllers (static => per process) */
 namespace
 {
   /** list of controllers suspected of failure */
-  SuspectedList suspected_controllers;
-  /** # of controllers that have fail since this driver has been created */
+  SuspectList suspected_controllers;
+  /** # of controllers that have fail since this process has been created */
   int controller_failure_number = 0;
 };
 
@@ -68,24 +68,98 @@
   controller_list.clear();
 }
 
-void AbstractControllerConnectPolicy::updateSuspectList()
+/*static*/
+void AbstractControllerConnectPolicy::updateSuspectList(int select_timeout /* 
= 20 */)
 {
+  wstring fctName(L"AbstractControllerConnectPolicy::updateSuspectList");
+  if (suspected_controllers.size() < 1)
+    return;
+
   LockScope scLs(&suspected_controllers_CS);
-  for (SuspectedList::const_iterator iter = suspected_controllers.begin();
+  //create set of fds for select function
+  fd_set writableControllers;
+  FD_ZERO(&writableControllers);
+  struct timeval tv;
+  tv.tv_sec = 0;
+  tv.tv_usec = select_timeout;
+  int fdMax = -1, retVal = 0;
+  for (SuspectList::iterator iter = suspected_controllers.begin();
       iter != suspected_controllers.end(); iter++)
   {
-    //TODO:
-    //connect
-    //select
-    //foreach writable (in select result) => remove from list
+    try
+    {
+      //connect socket and add it to the list of fds
+      if 
((*iter).pingSocketPtr->connectTo((*iter).controllerInfo.getHostName(),
+                                           
(*iter).controllerInfo.getHostPort()))
+      {
+        int sock = (*iter).pingSocketPtr->getFd();
+        if (sock > fdMax)
+          fdMax = sock;
+        FD_SET(sock, &writableControllers);
+      }
+    }
+    catch (...){}
+  }
+  retVal = select(fdMax+1, NULL, &writableControllers, NULL, &tv);
+  if (retVal == -1)
+  {
+    if (isErrorEnabled())
+    {
+      logError(fctName, L"Select returned error #"+toWString(errno));
+    }
+    return;
+  }
+  // if (retVal == 0) => timeout, no controller up again, do nothing
+  // If controllers are found up again, let's remove them from the list of
+  // suspects
+  if (retVal > 0)
+  {
+    // Anti-perfomant-but-very-sure way to remove up-again controllers from the
+    // list of suspects
+    bool mayHaveControllerLeftToRemove = true;
+    while (mayHaveControllerLeftToRemove)
+    {
+      mayHaveControllerLeftToRemove = false;
+      for (SuspectList::iterator iter = suspected_controllers.begin();
+          iter != suspected_controllers.end(); iter++)
+      {
+        if (FD_ISSET((*iter).pingSocketPtr->getFd(), &writableControllers))
+        {
+          if (isDebugEnabled())
+          {
+            logDebug(fctName, L"Controller "
+                + static_cast<wstring>((*iter).controllerInfo)
+                + L" is up again. Removing it from the list of suspects (list 
size="
+                + toWString(suspected_controllers.size())+L")");
+          }
+          //send ping and close socket
+          (*iter).pingSocketPtr->writeJavaInt(Ping);
+          // For performance, we could:
+          //1.delete (*iter).pingSocketPtr;
+          //2.suspected_controllers.erase(iter);
+          // But we don't need perf in this part of code, so let's factorize
+          // and use our dedicated function
+          removeControllerFromSuspectList((*iter).controllerInfo);
+          mayHaveControllerLeftToRemove = true;
+          break;
+        }
+      }
+    }
   }
 }
 
+/*static*/
 bool AbstractControllerConnectPolicy::isSuspectedOfFailure(
     const ControllerInfo& controllerInfo)
 {
-  SuspectedList::const_iterator iter = 
suspected_controllers.find(controllerInfo);
-  return iter != suspected_controllers.end();
+  // Just compare controller info with each item in the whole list
+  for (SuspectList::const_iterator iter = suspected_controllers.begin();
+      iter != suspected_controllers.end(); iter++)
+  {
+    if ((*iter).controllerInfo == controllerInfo)
+      return true;
+  }
+  return false;
 }
 
 void AbstractControllerConnectPolicy::suspectControllerOfFailure(
@@ -97,42 +171,62 @@
   if (isSuspectedOfFailure(controllerInfo))
     return;
 
-  // Check that the controllerInfo is correct and add it to the list
+  // Add it to the list
   for (size_t i = 0; i < controller_list.size(); i++)
   {
-    ControllerInfo controller = controller_list[i];
-    if (controller == controllerInfo)
+    if (controllerInfo == controller_list[i])
     {
-      // prepare the ping operation to this controller
-      JavaSocket newSocket;
-      newSocket.create(false); //create non blocking
       LockScope scLs(&suspected_controllers_CS);
-      // assign a failure number to this controller
+      //just for reporting (in the future ?)
       controller_failure_number++;
-      //controllerInfo.setFailureNumber(controller_failure_number);
-      //add the controller and its prepared socket to the list
-      suspected_controllers[controllerInfo] = newSocket;
+      SuspectController newSuspect;
+      newSuspect.controllerInfo = controllerInfo;
+      // creates and prepares a new socket for pinging the controller
+      newSuspect.pingSocketPtr = new JavaSocket();
+      newSuspect.pingSocketPtr->create(false); //create non blocking
+      //add the controller and its socket to the list
+      suspected_controllers.push_back(newSuspect);
       if (isDebugEnabled())
-        logDebug(fctName, L"Controller " + (wstring)controllerInfo
-            + L" is now suspected of failure (#");
+        logDebug(fctName, L"Controller " + static_cast<wstring>(controllerInfo)
+            + L" is now suspected of failure (list size="
+            + toWString(suspected_controllers.size())+L")");
       return;
     }
   }
 }
 
+/*static*/
 void AbstractControllerConnectPolicy::removeControllerFromSuspectList(
-    const ControllerInfo& controller)
+    const ControllerInfo& controllerInfo)
 {
   wstring 
fctName(L"AbstractControllerConnectPolicy::removeControllerFromSuspectList");
   LockScope scLs(&suspected_controllers_CS);
-  suspected_controllers.erase(controller);
-  if (isDebugEnabled())
+
+  for (SuspectList::iterator iter = suspected_controllers.begin();
+      iter != suspected_controllers.end(); iter++)
+  {
+    if ((*iter).controllerInfo == controllerInfo)
+    {
+      // found it
+      delete (*iter).pingSocketPtr; //free the ping socket
+      suspected_controllers.erase(iter);
+      if (isDebugEnabled())
+      {
+        logDebug(fctName, L"Controller " + (wstring)controllerInfo
+            + L" has been removed from suspect list (list size="
+            + toWString(suspected_controllers.size())+L")");
+      }
+      return;
+    }
+  }
+  if (isWarningEnabled())
   {
-    logDebug(fctName, L"Controller " + (wstring)controller
-        + L" is removed from suspect list");
+    logWarning(fctName, L"Controller " + (wstring)controllerInfo
+        + L" is not (anymore?) in the suspect list");
   }
 }
 
+/*static*/
 int AbstractControllerConnectPolicy::getNumberOfFailures()
 {
   return controller_failure_number;
@@ -158,15 +252,21 @@
   LockScope ls(&policy_CS);
   {
     LockScope scLs(&suspected_controllers_CS);
-    if (suspected_controllers.size() == controller_list.size())
-      throw NoMoreControllerException(L"All "
-                                      + toWString(suspected_controllers.size())
-                                      + L" controllers down");
+    unsigned int testedControllers = 0;
     do
     {
       index = (index + 1) % controller_list.size();
+      testedControllers++;
+    }
+    while (isSuspectedOfFailure(controller_list[index])
+        && testedControllers <= controller_list.size());
+    // if there are on more controllers up => exception
+    if (testedControllers > controller_list.size())
+    {
+      throw NoMoreControllerException(L"All "
+                                      + toWString(suspected_controllers.size())
+                                      + L" controllers down");
     }
-    while (isSuspectedOfFailure(controller_list[index]));
   }
   if (isDebugEnabled())
     logDebug(fctName, L"Selected controller[" + toWString(index) + L"]:"