Lock system implementation for concurrent access to a Virtual Instrument

Benjamin Renard
1 parent 3c7c7fa7
Showing 9 changed files with 363 additions and 5 deletions Show diff stats
CMakeLists.txt
README
scripts/StartServer.in
src/INCLUDE/DD_comm.h
src/SERVER/CMakeLists.txt
src/SERVER/DD_CacheLock.c
src/SERVER/DD_Server.c
tests/CMakeLists.txt
tests/testParallel.c
@@ -33,6 +33,7 @@ configure_file (
 MESSAGE( STATUS "Build DD_Server Project" )
 add_subdirectory(src/CLIENT)
 add_subdirectory(src/SERVER)
+add_subdirectory(tests)
  
 install(FILES "scripts/StartServer" DESTINATION . PERMISSIONS OWNER_READ OWNER_WRITE OWNER_EXECUTE GROUP_READ GROUP_EXECUTE WORLD_READ WORLD_EXECUTE)
 install(DIRECTORY "src/INCLUDE/" DESTINATION include)
 1. Set NETCDF_ROOT to use a specific netcdf installation directory :
-> export NETCDF_ROOT="/home/benjamin/AMDA-GIT/AMDA_COTS/netcdf/install"
+> export NETCDF_ROOT="/usr/lib/"
  
 2. Make the build directory :
 > cmake -E make_directory build
@@ -2,8 +2,8 @@
 #-------------------------------------------------------------------
 #                    StartServer
 #-------------------------------------------------------------------
-DDBASE=@CMAKE_INSTALL_PREFIX@/DDBASE/DATA
-DDPATH=@CMAKE_INSTALL_PREFIX@/DDBASE
+DDBASE=/home/budnik/AMDA-NG.core/DDBASE/DATA
+DDPATH=/home/budnik/AMDA-NG.core/DDBASE
 DDLIB=@CMAKE_INSTALL_PREFIX@/lib
 DDBASEBIN=@CMAKE_INSTALL_PREFIX@/bin
 LD_LIBRARY_PATH=$DDLIB/:@NETCDFLIB_DIR@:@USRLIB_DIR@
@@ -273,6 +273,27 @@ enum LockStatus {NOLOCK, LOCKED, LOCKREMOVED};
  
 /*------------------ Function for Server ---------------------*/
 extern int OpenInstr(char *InstrName);
+
+/*
+ * Init shared memory for VI cache lock system
+ */
+extern void CacheLock_Init();
+
+/*
+ * Free shared memory used for VI cache lock system
+ */
+extern void CacheLock_Free();
+
+/*
+ * Lock cache for a specific VI
+ */
+extern void CacheLock_Lock(char* VIID);
+
+/*
+ * Unlock cache for a specific VI
+ */
+extern void CacheLock_Unlock(char* VIID);
+
 /*
  * Open Virtual instrument by name and returns the ID
  * Returns negative value in case of error (see DD.h)or OK
@@ -21,6 +21,7 @@ target_link_libraries(
   DD_Client
   ${NETCDFLIBRARY}
   ${CRYPT_LIBRARY}
+  mpi
 )
  
 install (TARGETS DD_Server DESTINATION bin)
@@ -0,0 +1,143 @@
+/*=====================================================================
+ *                DD SYSTEM base package
+ *                  DD_Server library
+ *                    DD_CacheLock.c
+ *                      V.1.0
+ * Last revision:
+ *   May 29 2015: Version 1.0. First implementation of a lock system for concurrent access to a virtual instrument
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <pthread.h>
+
+typedef struct
+{
+  char VIID[30];
+  pthread_mutex_t mutex;
+} CacheLockVI;
+
+#define NB_CACHELOCK_VI 30
+typedef struct
+{
+  CacheLockVI cacheLockVI[NB_CACHELOCK_VI];
+  pthread_mutex_t mutex;
+} CacheLockData;
+
+extern int Verbose;
+
+static CacheLockData* cachelock_data = NULL;
+
+/*
+ * Init the shared memory used by the lock system
+ */
+void CacheLock_Init()
+{
+   if (cachelock_data != NULL)
+      return;
+
+   if(Verbose) fprintf(stderr,"CacheLock_Init\n");
+
+   // place our shared data in shared memory
+   int prot = PROT_READ | PROT_WRITE;
+   int flags = MAP_SHARED | MAP_ANONYMOUS;
+   cachelock_data = mmap(NULL, sizeof(CacheLockData), prot, flags, -1, 0);
+
+   // initialise mutex so it works properly in shared memory
+
+   // global mutex
+   pthread_mutexattr_t attr;
+   pthread_mutexattr_init(&attr);
+   pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
+   pthread_mutex_init(&cachelock_data->mutex, &attr);
+
+   int i;
+   for (i = 0; i < NB_CACHELOCK_VI; ++i)
+   {
+      // virtual instrument mutex
+      pthread_mutexattr_t attr;
+      pthread_mutexattr_init(&attr);
+      pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
+      pthread_mutex_init(&cachelock_data->cacheLockVI[i].mutex, &attr);
+   }
+}
+
+/*
+ * Free the shared memory used by the lock system
+ */
+void CacheLock_Free()
+{
+   if(Verbose) fprintf(stderr,"CacheLock_Free\n");
+   if (cachelock_data != NULL)
+     munmap(cachelock_data, sizeof(CacheLockData));
+   cachelock_data = NULL;
+}
+
+/*
+ * Get and lock the access to a virtual instrument
+ */
+void CacheLock_Lock(char* VIID)
+{
+   if (cachelock_data == NULL)
+      return;
+
+   int i;
+   pthread_mutex_lock(&cachelock_data->mutex);
+   for (i = 0; i < NB_CACHELOCK_VI; ++i)
+   {
+      if (strcmp(VIID, cachelock_data->cacheLockVI[i].VIID) == 0)
+      {
+        //A cache lock already exist for this VI
+        //Try to get access
+        pthread_mutex_lock(&cachelock_data->cacheLockVI[i].mutex);
+        if(Verbose) fprintf(stderr,"CacheLock_Lock. %s is locked\n", VIID);
+        pthread_mutex_unlock(&cachelock_data->mutex);
+        return;
+      }
+   }
+   pthread_mutex_unlock(&cachelock_data->mutex);
+
+   //No cache lock exist for this VI
+   //Try to find an available cache lock
+   while (1)
+   {
+     pthread_mutex_lock(&cachelock_data->mutex);
+     for (i = 0; i < NB_CACHELOCK_VI; ++i)
+     {
+        if (pthread_mutex_trylock(&cachelock_data->cacheLockVI[i].mutex) == 0)
+        {
+           //This cache lock is now used for this VIID
+           strcpy(cachelock_data->cacheLockVI[i].VIID, VIID);
+           if(Verbose) fprintf(stderr,"CacheLock_Lock. %s is locked\n", VIID);
+           pthread_mutex_unlock(&cachelock_data->mutex);
+           return;
+        }
+     }
+     //No available cache lock for the moment
+     if(Verbose) fprintf(stderr,"CacheLock_Lock. Cannot find an available cache lock for %s => Wait and retry later\n", VIID);
+     pthread_mutex_unlock(&cachelock_data->mutex);
+     sleep(2);
+   }
+}
+
+/*
+ * Unlock the access to a virtual instrument
+ */
+void CacheLock_Unlock(char* VIID)
+{
+   if (cachelock_data == NULL)
+      return;
+
+   int i;
+   for (i = 0; i < NB_CACHELOCK_VI; ++i)
+   {
+      if (strcmp(VIID, cachelock_data->cacheLockVI[i].VIID) == 0)
+      {
+        if(Verbose) fprintf(stderr,"CacheLock_Unlock. %s is unlocked\n", VIID);
+        pthread_mutex_unlock(&cachelock_data->cacheLockVI[i].mutex);
+        return;
+      }
+   }
+   if(Verbose) fprintf(stderr,"CacheLock_Unlock. Cannot unlock %s\n", VIID);
+}
@@ -179,7 +179,7 @@ int SendToNet(int SocketID, char *buff, int length)
 }
  
 /*------------------ PUTHEADER -----------------------------*
-/*
+ *
  * Encode end send Reply Header
  */
 int PutHeader(int a1, int a2, int a3,int SocketID)
@@ -358,7 +358,7 @@ void Serv(int SocketID)
    static char *NewVIName = NULL;    /* Full name of VI (data set to generate )*/
    static char *RemSetID =  NULL;    /* ID of remote Data Set */
    static char *RemBaseName = NULL;  /* Base Name of remote WEB service */
-   static char Command[MAXCOMMANDL];  /*Memory for command in system()
+   static char Command[MAXCOMMANDL];  /*Memory for command in system() */
    /*--------------------------------------------------*/
  
    /*---------------------------------------------------
@@ -578,7 +578,9 @@ void Serv(int SocketID)
                  /*------ Call data until LastPacket Flag (OK) returns --------*/
                  else do
                  {
+                      CacheLock_Lock(DD_Var[id]->InstrName);
                       err = GetMultiData(id,VarSize,VarNames,ReqTime,BackFlag);
+                      CacheLock_Unlock(DD_Var[id]->InstrName);
                       if(err >= 0)
                       {
 			serr = SendDataPacket(DD_Var[id],SocketID,DD_Var[id]->LastPacketFlag); /* OK, MOREDATA, MOREDELAY */
@@ -661,6 +663,8 @@ int main(int argc, char **argv)
    else if(((argv[1])[1] == 'v') || ((argv[1])[1] == 'V')) Verbose = 1;
    else Verbose = 0;
  
+   CacheLock_Init();
+
    if((ListenerID = SetListener()) < 0)
    {
      fprintf(stderr,"DD_server: I cant set Listener Socket\n");
@@ -697,5 +701,8 @@ int main(int argc, char **argv)
    /* Unreachable point */
  
    close(ListenerID);
+
+   CacheLock_Free();
+
    return 0;
 }
@@ -0,0 +1,23 @@
+
+PROJECT(testParallel)
+
+include_directories(
+  ${CMAKE_HOME_DIRECTORY}/src/INCLUDE/
+)
+
+#Configuration de l'exécutable
+file(
+        GLOB_RECURSE
+        source_files
+        ./*
+)
+  
+ADD_EXECUTABLE (testParallel ${source_files} )
+
+target_link_libraries(
+  testParallel
+  ${CMAKE_THREAD_LIBS_INIT}
+  DD_Client
+)
+
+install (TARGETS testParallel DESTINATION tests)
@@ -0,0 +1,162 @@
+/*=============================================================
+ * testParallel.c
+ * Test concurrent access of DD_Server
+ *  May 2015, V.1.0, Renard
+ *=============================================================*/
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <DD.h>
+#include <time.h>
+#include <string.h>
+#include <sys/time.h>
+
+//Structrure used to define a request
+typedef struct {
+   char ViName[100];
+   char StartTime[17];
+   char TimeInt[17];
+   char TimeName[20];
+   char ParName[20];
+} RequestData;
+
+//Number of requests to run
+#define NB_REQUESTS 60
+
+//Array of all requests data
+RequestData requestDataList[NB_REQUESTS];
+
+/*-------------------------*/
+#define IMF_VI    "ace:imf:all\0"
+#define IMF_PAR   "IMF\0"
+#define IMF_TIME  "Time\0"
+#define IMF_START "2008000000000000\0"
+#define IMF_TI    "0000030000000000\0"
+
+#define DST_VI    "ground:based:dst\0"
+#define DST_PAR   "DST\0"
+#define DST_TIME  "Time\0"
+#define DST_START "2008000000000000\0"
+#define DST_TI    "0000060000000000\0"
+
+#define MESMAG_VI    "mes:mag:orb\0"
+#define MESMAG_PAR   "B_MSO\0"
+#define MESMAG_TIME  "Time\0"
+#define MESMAG_START "2012000000000000\0"
+#define MESMAG_TI    "0000004000000000\0"
+/*-------------------------*/
+
+/*
+ * Init requests data
+ */
+void initRequestDataList()
+{
+   memset(&requestDataList, 0, NB_REQUESTS*sizeof(RequestData));
+
+   int i;
+   for (i = 0; i < NB_REQUESTS; ++i)
+   {
+      if (i%3 == 0)
+      {
+         strcpy(requestDataList[i].ViName,MESMAG_VI);
+         strcpy(requestDataList[i].StartTime,MESMAG_START);
+         strcpy(requestDataList[i].TimeInt,MESMAG_TI);
+         strcpy(requestDataList[i].TimeName,MESMAG_TIME);
+         strcpy(requestDataList[i].ParName,MESMAG_PAR);
+      }
+      else if (i%2 == 0)
+      {
+         strcpy(requestDataList[i].ViName,DST_VI);
+         strcpy(requestDataList[i].StartTime,DST_START);
+         strcpy(requestDataList[i].TimeInt,DST_TI);
+         strcpy(requestDataList[i].TimeName,DST_TIME);
+         strcpy(requestDataList[i].ParName,DST_PAR);
+      }
+      else
+      {
+
+         strcpy(requestDataList[i].ViName,IMF_VI);
+         strcpy(requestDataList[i].StartTime,IMF_START);
+         strcpy(requestDataList[i].TimeInt,IMF_TI);
+         strcpy(requestDataList[i].TimeName,IMF_TIME);
+         strcpy(requestDataList[i].ParName,IMF_PAR);
+      }
+   }
+}
+
+/*
+ * Run a request
+ */
+int runRequest(char* ViName, char* StartTime, char* TimeInt, int NbPar, char** ParNames)
+{
+   int ID, error;
+   double RealTime;
+   DD_data_t *data;
+
+   ID = DD_SetVariable(ViName);
+   if(ID < 0)
+   {
+     error = DD_Close(99);
+     return 0;
+   }
+
+   error = DD_SetTimeInfo(ID, StartTime, &RealTime);   
+   if(error < 0)
+   {
+     error = DD_Close(ID);
+     return 0;
+   }
+
+   do
+   {
+      error = DD_GetMultiData(ID, NbPar, ParNames, TimeInt, &data, 1);
+      if(error < 0)
+      {
+         error = DD_Close(ID);
+         return 0;
+      }
+      if(error == MOREDELAY)
+      {
+         error = MOREDATA;
+      }
+   }
+   while(error == MOREDATA);
+
+   error = DD_Close(ID);
+
+   return 1;
+}
+
+/*
+ * Main
+ */
+int main()
+{
+   initRequestDataList();
+
+   pid_t pid;
+   int crtRequest = 0;
+
+   do {
+      pid = fork();
+      if (pid == 0)
+      {
+         char *ParNames[2] = {(char *)requestDataList[crtRequest].TimeName, (char *)requestDataList[crtRequest].ParName};
+         printf("Run Request %s %s %s %s\n",requestDataList[crtRequest].ViName,requestDataList[crtRequest].ParName,requestDataList[crtRequest].StartTime,requestDataList[crtRequest].TimeInt);
+         if (!runRequest(requestDataList[crtRequest].ViName, requestDataList[crtRequest].StartTime, requestDataList[crtRequest].TimeInt, 2, ParNames))
+           printf("Error detected %s %s %s %s\n",requestDataList[crtRequest].ViName,requestDataList[crtRequest].ParName,requestDataList[crtRequest].StartTime,requestDataList[crtRequest].TimeInt);
+         else
+           printf("OK %s %s %s %s\n",requestDataList[crtRequest].ViName,requestDataList[crtRequest].ParName,requestDataList[crtRequest].StartTime,requestDataList[crtRequest].TimeInt);
+      }
+      else
+      {
+        ++crtRequest;
+        if (crtRequest == NB_REQUESTS)
+        {
+           wait(0);
+        }
+      }
+   } while ((pid != 0) && (crtRequest < NB_REQUESTS));
+
+   return 1;
+}