Changeset 507

Timestamp:

Oct 8, 2014, 11:38:12 AM (10 years ago)

Author:

djay

Message:

Add concurrency gesture for background services. Fix tickets #102 and #103.

Location:

trunk/zoo-project/zoo-kernel

Files:

• service.h (modified) (3 diffs)
• service_internal.c (modified) (17 diffs)
• service_internal.h (modified) (3 diffs)
• ulinet.h (modified) (2 diffs)
• zoo_service_loader.c (modified) (6 diffs)

trunk/zoo-project/zoo-kernel/service.h

@@ -40 +40 @@
 #define zOpen _open
 #define zWrite _write
+#define zSleep Sleep
 #else
 #define zStrdup strdup
@@ -45 +46 @@
 #define zOpen open
 #define zWrite write
+#define zSleep sleep
 #endif 
 
@@ -68 +70 @@
 #define false -1
 #endif
-#else
-  //#include <stdbool.h>
 #endif
 

trunk/zoo-project/zoo-kernel/service_internal.c

@@ -83 +83 @@
 }
 
+char* _getStatus(maps* conf,int pid){
+  char lid[1024];
+  sprintf(lid,"%d",pid);
+  setMapInMaps(conf,"lenv","lid",lid);
+  semid lockid=getShmLockId(conf,1);
+  if(
+#ifdef WIN32
+     lockid==NULL
+#else
+     lockid<0
+#endif
+     ){
+    char tmp[3];
+    sprintf(tmp,"%d",ZOO_LOCK_CREATE_FAILED);
+    return tmp;
+  }
+  if(lockShm(lockid)<0){
+    fprintf(stderr,"%s %d\n",__FILE__,__LINE__);
+    fflush(stderr);
+    char tmp[3];
+    sprintf(tmp,"%d",ZOO_LOCK_ACQUIRE_FAILED);
+    return tmp;
+  }
+  char *tmp=getStatus(pid);
+  unlockShm(lockid);
+  if(tmp==NULL || strncmp(tmp,"-1",2)==0){
+    removeShmLock(conf,1);
+  }
+  return tmp;
+}
+
 #ifdef WIN32
 
@@ -92 +123 @@
 
 #define SHMEMSIZE 4096
+
+char* getKeyValue(maps* conf){
+  map *tmpMap=getMapFromMaps(conf,"lenv","lid");
+  if(tmpMap==NULL)
+    tmpMap=getMapFromMaps(conf,"lenv","sid");
+  char* key="-1";
+  if(tmpMap!=NULL){
+    key=(char*)malloc((strlen(tmpMap->value)+9)*sizeof(char));
+    sprintf(key,"zoo_sem_%s",tmpMap->value);
+  }
+  return key;
+}
+
+
+semid getShmLockId(maps* conf, int nsems){
+    semid sem_id;
+    char* key=getKeyValue(conf);
+    
+    sem_id = CreateSemaphore( NULL, nsems, nsems+1, key);
+    if(sem_id==NULL){
+      if(strncmp(key,"-1",2)!=0)
+        free(key);
+#ifdef DEBUG
+      fprintf(stderr,"Semaphore failed to create ! %s\n",GetLastError());
+#endif
+      return NULL;
+    }
+#ifdef DEBUG
+    fprintf(stderr,"%s Accessed !\n",key);
+#endif
+    if(strncmp(key,"-1",2)!=0)
+      free(key);
+    return sem_id;
+}
+
+int removeShmLock(maps* conf, int nsems){
+  semid sem_id=getShmLockId(conf,1);
+  if (CloseHandle(sem_id) == 0) {
+    fprintf(stderr,"Unable to remove semaphore %s",GetLastError());
+    return -1;
+  }
+#ifdef DEBUG
+  fprintf(stderr,"%d Removed !\n",sem_id);
+#endif
+  return 0;
+}
+
+int lockShm(semid id){
+  DWORD dwWaitResult=WaitForSingleObject(id,INFINITE);
+  switch (dwWaitResult){
+    case WAIT_OBJECT_0:
+      return 0;
+      break;
+    case WAIT_TIMEOUT:
+      return -1;
+      break;
+    default:
+      return -2;
+      break;
+  }
+  return 0;
+}
+
+int unlockShm(semid id){
+  if(!ReleaseSemaphore(id,1,NULL)){
+    return -1;
+  }
+  return 0;
+}
 
 static LPVOID lpvMemG = NULL;      // pointer to shared memory
@@ -103 +203 @@
   map *tmpMap1;
   map *tmpMap=getMapFromMaps(conf,"lenv","sid");
+  semid lockid=getShmLockId(conf,1);
+  if(lockid==NULL){
+#ifdef DEBUG
+    fprintf(stderr,"Unable to create semaphore on line %d!! \n",__LINE__);
+#endif
+    return ZOO_LOCK_CREATE_FAILED;
+  }
+  if(lockShm(lockid)<0){
+#ifdef DEBUG
+    fprintf(stderr,"Unable to create semaphore on line %d!! \n",__LINE__);
+#endif
+    return ZOO_LOCK_ACQUIRE_FAILED;
+  }
+  
   if(hMapObjectG==NULL)
     hMapObjectG = CreateFileMapping( 
@@ -112 +226 @@
                                     TEXT(tmpMap->value));   // name of map object
   if (hMapObjectG == NULL){
-    fprintf(stderr,"Unable to create share memory segment %s !! \n",tmpMap->value);
+#ifdef DEBUG
+    fprintf(stderr,"Unable to create shared memory segment %d !! \n",GetLastError());
+#endif
     return -2;
   }
@@ -124 +240 @@
                             0);             // default: map entire file
   if (lpvMemG == NULL){
+#ifdef DEBUG
     fprintf(stderr,"Unable to create or access the shared memory segment %s !! \n",tmpMap->value);
+#endif
     return -1;
   } 
@@ -139 +257 @@
   *lpszTmp++ = '\0';
   free(final_string);
+  unlockShm(lockid);
   return 0;
 }
@@ -149 +268 @@
   HANDLE hMapObject = NULL;
   BOOL fIgnore,fInit;
-  char tmp[100];
-  sprintf(tmp,"%i",pid);
+  char tmp[1024];
+  sprintf(tmp,"%d",pid);
   if(hMapObject==NULL)
     hMapObject = CreateFileMapping( 
@@ -159 +278 @@
                                    4096,                   // size: low 32-bits
                                    TEXT(tmp));   // name of map object
-  if (hMapObject == NULL) 
-    return FALSE;
+  if (hMapObject == NULL){
+#ifdef DEBUG
+    fprintf(stderr,"ERROR on line %d\n",__LINE__);
+#endif
+    return "-1";
+  }
   if((GetLastError() != ERROR_ALREADY_EXISTS)){
+#ifdef DEBUG
+    fprintf(stderr,"ERROR on line %d\n",__LINE__);
+    fprintf(stderr,"READING STRING S %s\n",GetLastError());
+#endif
     fIgnore = UnmapViewOfFile(lpvMem); 
     fIgnore = CloseHandle(hMapObject);
@@ -174 +301 @@
                            0,              // low offset:   beginning
                            0);             // default: map entire file
-  if (lpvMem == NULL) 
+  if (lpvMem == NULL){
+#ifdef DEBUG
+    fprintf(stderr,"READING STRING S %d\n",__LINE__);
+    fprintf(stderr,"READING STRING S %s\n",GetLastError());
+#endif
     return "-1"; 
+  }
   lpszTmp = (LPWSTR) lpvMem;
   while (*lpszTmp){
@@ -183 +315 @@
     i++;
   }
-#ifdef DEBUG
-  fprintf(stderr,"READING STRING S %s\n",lpszBuf);
-#endif
   return (char*)lpszBuf;
 }
@@ -194 +323 @@
   fIgnore = CloseHandle(hMapObjectG);
 }
+
 #else
+
+#define MAX_RETRIES 10
+
+int getKeyValue(maps* conf){
+  map *tmpMap=getMapFromMaps(conf,"lenv","lid");
+  if(tmpMap==NULL)
+    tmpMap=getMapFromMaps(conf,"lenv","sid");
+  int key=-1;
+  if(tmpMap!=NULL)
+    key=atoi(tmpMap->value);
+  return key;
+}
+
+int getShmLockId(maps* conf, int nsems){
+    int i;
+    union semun arg;
+    struct semid_ds buf;
+    struct sembuf sb;
+    semid sem_id;
+    int key=getKeyValue(conf);
+    
+    sem_id = semget(key, nsems, IPC_CREAT | IPC_EXCL | 0666);
+
+    if (sem_id >= 0) { /* we got it first */
+        sb.sem_op = 1; 
+        sb.sem_flg = 0;
+        arg.val=1;
+        for(sb.sem_num = 0; sb.sem_num < nsems; sb.sem_num++) { 
+            /* do a semop() to "free" the semaphores. */
+            /* this sets the sem_otime field, as needed below. */
+            if (semop(sem_id, &sb, 1) == -1) {
+                int e = errno;
+                semctl(sem_id, 0, IPC_RMID); /* clean up */
+                errno = e;
+                return -1; /* error, check errno */
+            }
+        }
+    } else if (errno == EEXIST) { /* someone else got it first */
+        int ready = 0;
+
+        sem_id = semget(key, nsems, 0); /* get the id */
+        if (sem_id < 0) return sem_id; /* error, check errno */
+
+        /* wait for other process to initialize the semaphore: */
+        arg.buf = &buf;
+        for(i = 0; i < MAX_RETRIES && !ready; i++) {
+            semctl(sem_id, nsems-1, IPC_STAT, arg);
+            if (arg.buf->sem_otime != 0) {
+#ifdef DEBUG
+              fprintf(stderr,"Semaphore acquired ...\n");
+#endif
+              ready = 1;
+            } else {
+#ifdef DEBUG
+              fprintf(stderr,"Retry to access the semaphore later ...\n");
+#endif
+              sleep(1);
+            }
+        }
+        errno = NULL;
+        if (!ready) {
+#ifdef DEBUG
+          fprintf(stderr,"Unable to access the semaphore ...\n");
+#endif
+          errno = ETIME;
+          return -1;
+        }
+    } else {
+        return sem_id; /* error, check errno */
+    }
+#ifdef DEBUG
+    fprintf(stderr,"%d Created !\n",sem_id);
+#endif
+    return sem_id;
+}
+
+int removeShmLock(maps* conf, int nsems){
+  union semun arg;
+  int sem_id=getShmLockId(conf,nsems);
+  if (semctl(sem_id, 0, IPC_RMID, arg) == -1) {
+    perror("semctl");
+    return -1;
+  }
+  return 0;
+}
+
+int lockShm(int id){
+  struct sembuf sb;
+  int i;
+  sb.sem_num = 0;
+  sb.sem_op = -1;  /* set to allocate resource */
+  sb.sem_flg = SEM_UNDO;
+  if (semop(id, &sb, 1) == -1){
+    perror("semop");
+    return -1;
+  }
+  return 0;
+}
+
+int unlockShm(int id){
+  struct sembuf sb;
+  sb.sem_num = 0;
+  sb.sem_op = 1;  /* free resource */
+  sb.sem_flg = SEM_UNDO;
+  if (semop(id, &sb, 1) == -1) {
+    perror("semop");
+    return -1;
+  }
+  return 0;
+}
 
 void unhandleStatus(maps *conf){
@@ -223 +463 @@
 int _updateStatus(maps *conf){
   int shmid;
-  key_t key;
   char *shm,*s,*s1;
   map *tmpMap=NULL;
-  tmpMap=getMapFromMaps(conf,"lenv","sid");
-  if(tmpMap!=NULL){
-    key=atoi(tmpMap->value);
+  key_t key=getKeyValue(conf);
+  if(key!=-1){
+    semid lockid=getShmLockId(conf,1);
+    if(lockid<0)
+      return ZOO_LOCK_CREATE_FAILED;
+    if(lockShm(lockid)<0){
+      return ZOO_LOCK_ACQUIRE_FAILED;
+    }
     if ((shmid = shmget(key, SHMSZ, IPC_CREAT | 0666)) < 0) {
 #ifdef DEBUG
       fprintf(stderr,"shmget failed to create new Shared memory segment\n");
 #endif
+      unlockShm(lockid);
       return -2;
     }else{
@@ -239 +484 @@
         fprintf(stderr,"shmat failed to update value\n");
 #endif
+        unlockShm(lockid);
         return -1;
       }
@@ -255 +501 @@
         *s1=NULL;
         shmdt((void *)shm);
+        unlockShm(lockid);
       }
     }
@@ -1112 +1359 @@
               if(nci0==0){
                 nc7 = xmlNewNode(ns_ows, BAD_CAST "MinimumValue");
-                int nci=1;
-                for(nci=1;nci<strlen(pToken);nci++){
-                  tmpStr[nci-1]=pToken[nci];
-                }
-                }else{
+                strncpy( tmpStr, pToken+1, strlen(pToken)-1 );
+                tmpStr[strlen(pToken)-1] = '\0';
+              }else{
                 nc7 = xmlNewNode(ns_ows, BAD_CAST "MaximumValue");
-                int nci=0;
-                for(nci=0;nci<strlen(pToken)-1;nci++){
-                  tmpStr[nci]=pToken[nci];
-                }
+                const char* bkt;
+                if ( ( bkt = strchr(pToken, '[') ) != NULL || ( bkt = strchr(pToken, ']') ) != NULL ){
+                    strncpy( tmpStr, pToken, bkt - pToken );
+                    tmpStr[bkt - pToken] = '\0';
+                  }
               }
               xmlAddChild(nc7,xmlNewText(BAD_CAST tmpStr));
@@ -1179 +1425 @@
             xmlAddChild(nc6,nc8);
             _tmp0=e->supported;
-            if(getMap(_tmp0->content,"range")!=NULL ||
-               getMap(_tmp0->content,"rangeMin")!=NULL ||
-               getMap(_tmp0->content,"rangeMax")!=NULL ||
-               getMap(_tmp0->content,"rangeClosure")!=NULL ){
+            if(_tmp0!=NULL &&
+               (getMap(_tmp0->content,"range")!=NULL ||
+                getMap(_tmp0->content,"rangeMin")!=NULL ||
+                getMap(_tmp0->content,"rangeMax")!=NULL ||
+                getMap(_tmp0->content,"rangeClosure")!=NULL )){
               tmp1=_tmp0->content;
               goto doRange;
@@ -1650 +1897 @@
   xmlDocSetRootElement(doc, nr);
 
-  if(hasStoredExecuteResponse==true){
-    /* We need to write the ExecuteResponse Document somewhere */
-    FILE* output=fopen(stored_path,"w");
-    if(output==NULL){
-      /* If the file cannot be created return an ExceptionReport */
-      char tmpMsg[1024];
-      sprintf(tmpMsg,_("Unable to create the file : \"%s\" for storing the ExecuteResponse."),stored_path);
-      map * errormap = createMap("text",tmpMsg);
-      addToMap(errormap,"code", "InternalError");
-      printExceptionReportResponse(m,errormap);
-      freeMap(&errormap);
-      free(errormap);
-      xmlFreeDoc(doc);
-      xmlCleanupParser();
-      zooXmlCleanupNs();
+  if(hasStoredExecuteResponse==true && status!=SERVICE_STARTED){
+    semid lid=getShmLockId(m,1);
+    if(lid<0)
       return;
-    }
-    xmlChar *xmlbuff;
-    int buffersize;
-    xmlDocDumpFormatMemoryEnc(doc, &xmlbuff, &buffersize, "UTF-8", 1);
-    fwrite(xmlbuff,1,xmlStrlen(xmlbuff)*sizeof(char),output);
-    xmlFree(xmlbuff);
-    fclose(output);
+    else{
+#ifdef DEBUG
+      fprintf(stderr,"LOCK %s %d !\n",__FILE__,__LINE__);
+#endif
+      lockShm(lid);
+      /* We need to write the ExecuteResponse Document somewhere */
+      FILE* output=fopen(stored_path,"w");
+      if(output==NULL){
+        /* If the file cannot be created return an ExceptionReport */
+        char tmpMsg[1024];
+        sprintf(tmpMsg,_("Unable to create the file : \"%s\" for storing the ExecuteResponse."),stored_path);
+        map * errormap = createMap("text",tmpMsg);
+        addToMap(errormap,"code", "InternalError");
+        printExceptionReportResponse(m,errormap);
+        freeMap(&errormap);
+        free(errormap);
+        xmlFreeDoc(doc);
+        xmlCleanupParser();
+        zooXmlCleanupNs();
+        unlockShm(lid);
+        return;
+      }
+      xmlChar *xmlbuff;
+      int buffersize;
+      xmlDocDumpFormatMemoryEnc(doc, &xmlbuff, &buffersize, "UTF-8", 1);
+      fwrite(xmlbuff,1,xmlStrlen(xmlbuff)*sizeof(char),output);
+      xmlFree(xmlbuff);
+      fclose(output);
+#ifdef DEBUG
+      fprintf(stderr,"UNLOCK %s %d !\n",__FILE__,__LINE__);
+#endif
+      unlockShm(lid);
+      map* test1=getMap(request,"status");
+      if(test1==NULL || strcasecmp(test1->value,"true")!=0){
+        removeShmLock(m,1);
+      }
+    }
   }
   printDocument(m,doc,pid);

trunk/zoo-project/zoo-kernel/service_internal.h

@@ -36 +36 @@
 #define _ss(String) dgettext ("zoo-services",String)
 
+#define ZOO_LOCK_CREATE_FAILED -4
+#define ZOO_LOCK_ACQUIRE_FAILED -5
+#define ZOO_LOCK_RELEASE_FAILED -6
+
 #include <sys/stat.h>
 #include <sys/types.h>
@@ -42 +46 @@
 #include <sys/ipc.h>
 #include <sys/shm.h>
+#include <sys/sem.h>
 #else
 #include <direct.h>
@@ -99 +104 @@
   void unhandleStatus(maps*);
   int _updateStatus(maps*);
+  char* _getStatus(maps*,int);
   char* getStatus(int);
+  int removeShmLock(maps*, int);
+#ifndef WIN32
+#define semid int
+#else
+#define semid HANDLE
+#endif
+  semid getShmLockId(maps*,int);
+  int lockShm(semid);
+  int unlockShm(semid);
 
 #ifdef USE_JS

trunk/zoo-project/zoo-kernel/ulinet.h

@@ -27 +27 @@
 
 #include "fcgi_stdio.h"
-#include "service.h"
 #include <stdlib.h>
 #include <fcntl.h>
@@ -44 +43 @@
 #include "jsapi.h"
 #endif
+#ifndef bool
+#define bool int
+#endif
+#ifndef true
+#define true 1
+#define false -1
+#endif
 
-#define MAX_REQ 100
+#define MAX_REQ 50
 
 #ifdef _ULINET

trunk/zoo-project/zoo-kernel/zoo_service_loader.c

@@ -2629 +2629 @@
        * process answer to http client.
        */
+#ifndef WIN32
+      zSleep(1);
+#endif
       r_inputs=getMapFromMaps(m,"main","tmpPath");
       map* r_inputs1=getMap(s1->content,"ServiceProvider");
@@ -2641 +2644 @@
 #endif
       freopen(flog, "w+", stderr);
+      semid lid=getShmLockId(m,1);
+      fflush(stderr);
+      if(lid<0){
+        fprintf(stderr,"ERROR %s %d\n",__FILE__,__LINE__);
+        fflush(stderr);
+        return -1;
+      }
+      else{
+        if(lockShm(lid)<0){
+          fprintf(stderr,"ERROR %s %d\n",__FILE__,__LINE__);
+          fflush(stderr);
+          return -1;
+        }
+        fflush(stderr);
+      }
       f0=freopen(fbkp , "w+", stdout);
+      rewind(stdout);
 #ifndef WIN32
       fclose(stdin);
@@ -2654 +2673 @@
       printProcessResponse(m,request_inputs,cpid,s1,r_inputs1->value,SERVICE_STARTED,
                            request_input_real_format,request_output_real_format);
-#ifndef WIN32
       fflush(stdout);
-      rewind(stdout);
-#else
-#endif
+      unlockShm(lid);
+      fflush(stderr);
       fbkp1=(char*)malloc((strlen(r_inputs->value)+strlen(r_inputs1->value)+1024)*sizeof(char));
       sprintf(fbkp1,"%s/%s_final_%d.xml",r_inputs->value,r_inputs1->value,cpid);
@@ -2697 +2714 @@
     fclose(stdout);
     fclose(stderr);
-    unhandleStatus(m);
     /**
      * Dump back the final file fbkp1 to fbkp
@@ -2704 +2720 @@
     fclose(f1);
     FILE* f2=fopen(fbkp1,"rb");
+    semid lid=getShmLockId(m,1);
+    if(lid<0)
+      return -1;
+    lockShm(lid);
     FILE* f3=fopen(fbkp,"wb+");
     free(fbkp);
@@ -2714 +2734 @@
     fclose(f2);
     fclose(f3);
+    unlockShm(lid);
     unlink(fbkp1);
     free(fbkp1);
     free(tmps1);
+    unhandleStatus(m);
   }