Port multiple client SHM to Windows

library/DFProcess-linux-SHM.cpp

@@ -50,7 +50,6 @@ class SHMProcess::Private
         shm_ID = -1;
         window = NULL;
         attached = false;
-        suspended = false;
         identified = false;
         useYield = false;
         server_lock = -1;
@@ -71,14 +70,14 @@ class SHMProcess::Private
     int suspend_lock;
     int attachmentIdx;
     
-    bool locked;
+
     
     bool attached;
-    bool suspended;
+    bool locked;
     bool identified;
     bool useYield;
     
-    bool validate(char* exe_file, uint32_t pid, std::vector< memory_info* >& known_versions);
+    bool validate(std::vector< memory_info* >& known_versions);
     
     bool Aux_Core_Attach(bool & versionOK, pid_t & PID);
     //bool waitWhile (uint32_t state);
@@ -110,13 +109,11 @@ bool SHMProcess::Private::SetAndWait (uint32_t state)
             {
                 //detach the shared memory
                 shmdt(shm_addr);
-                attached = suspended = identified = false;
+                FreeLocks();
+                attached = locked = identified = false;
                 // we aren't the current process anymore
                 g_pProcess = NULL;
-                FreeLocks();
-                
                 throw Error::SHMServerDisappeared();
-                return false;
             }
             else
             {
@@ -146,14 +143,6 @@ bool SHMProcess::SetAndWait (uint32_t state)
     return d->SetAndWait(state);
 }
 
-/*
-// set SHM command.
-void SHMProcess::setCmd (uint32_t newstate)
-{
-    if(d->attached && d->suspended)
-        D_SHMCMD = newstate;
-};
-*/
 uint32_t OS_getAffinity()
 {
     cpu_set_t mask;
@@ -163,25 +152,6 @@ uint32_t OS_getAffinity()
     return affinity;
 }
 
-bool SHMProcess::Private::Aux_Core_Attach(bool & versionOK, pid_t & PID)
-{
-    if(!locked) return false;
-    
-    SHMDATA(coreattach)->cl_affinity = OS_getAffinity();
-    if(!SetAndWait(CORE_ATTACH)) return false;
-    /*
-    cerr <<"CORE_VERSION" << CORE_VERSION << endl;
-    cerr <<"server CORE_VERSION" << SHMDATA(coreattach)->sv_version << endl;
-    */
-    versionOK =( SHMDATA(coreattach)->sv_version == CORE_VERSION );
-    PID = SHMDATA(coreattach)->sv_PID;
-    useYield = SHMDATA(coreattach)->sv_useYield;
-    #ifdef DEBUG
-        if(useYield) cerr << "Using Yield!" << endl;
-    #endif
-    return true;
-}
-
 // test if we have client and server locks and the server is present
 bool SHMProcess::Private::AreLocksOk()
 {
@@ -215,7 +185,6 @@ void SHMProcess::Private::FreeLocks()
     {
         close(suspend_lock);
         locked = false;
-        suspended = false;
         suspend_lock = -1;
     }
 }
@@ -229,7 +198,7 @@ bool SHMProcess::Private::GetLocks()
     server_lock = open(name,O_WRONLY);
     if(server_lock == -1)
     {
-        cerr << "can't open sv lock" << endl;
+        // cerr << "can't open sv lock" << endl;
         return false;
     }
     
@@ -289,27 +258,6 @@ bool SHMProcess::Private::GetLocks()
 SHMProcess::SHMProcess(uint32_t PID, vector< memory_info* >& known_versions)
 : d(new Private())
 {
-    char exe_link_name [256];
-    char target_name[1024];
-    int target_result;
-    
-    /*
-     * Locate the segment.
-     */
-    if ((d->shm_ID = shmget(SHM_KEY + PID, /*SHM_ALL_CLIENTS*/SHM_SIZE, 0666)) < 0)
-    {
-        return;
-    }
-    
-    /*
-     * Attach the segment
-     */
-    /*
-    if ((d->shm_addr = (char *) shmat(d->shm_ID, NULL, 0)) == (char *) -1)
-    {
-        return;
-    }
-    */
     d->process_ID = PID;
     if(!attach())
     {
@@ -324,31 +272,15 @@ SHMProcess::SHMProcess(uint32_t PID, vector< memory_info* >& known_versions)
     {
         detach();
         throw Error::SHMAttachFailure();
-        return;
     }
     if(!bridgeOK)
     {
-        throw Error::SHMVersionMismatch();
-        detach();
-        
-        return;
-    }
-    
-    // find the binary
-    sprintf(exe_link_name,"/proc/%d/exe", d->process_ID);
-    target_result = readlink(exe_link_name, target_name, sizeof(target_name)-1);
-    if (target_result == -1)
-    {
-        perror("readlink");
         detach();
-        return;
+        throw Error::SHMVersionMismatch();
     }
-    // make sure we have a null terminated string...
-    // see http://www.opengroup.org/onlinepubs/000095399/functions/readlink.html
-    target_name[target_result] = 0;
     
     // try to identify the DF version (md5 the binary, compare with known versions)
-    d->validate(target_name, d->process_ID, known_versions);
+    d->validate(known_versions);
     d->window = new DFWindow(this);
     
     // detach
@@ -357,7 +289,7 @@ SHMProcess::SHMProcess(uint32_t PID, vector< memory_info* >& known_versions)
 
 bool SHMProcess::isSuspended()
 {
-    return d->suspended;
+    return d->locked;
 }
 bool SHMProcess::isAttached()
 {
@@ -369,11 +301,26 @@ bool SHMProcess::isIdentified()
     return d->identified;
 }
 
-bool SHMProcess::Private::validate(char * exe_file, uint32_t pid, vector <memory_info *> & known_versions)
+bool SHMProcess::Private::validate(vector <memory_info *> & known_versions)
 {
+    char exe_link_name [256];
+    char target_name[1024];
+    int target_result;
+    // find the binary
+    sprintf(exe_link_name,"/proc/%d/exe", process_ID);
+    target_result = readlink(exe_link_name, target_name, sizeof(target_name)-1);
+    if (target_result == -1)
+    {
+        perror("readlink");
+        return false;
+    }
+    // make sure we have a null terminated string...
+    // see http://www.opengroup.org/onlinepubs/000095399/functions/readlink.html
+    target_name[target_result] = 0;
+    
     md5wrapper md5;
     // get hash of the running DF process
-    string hash = md5.getHashFromFile(exe_file);
+    string hash = md5.getHashFromFile(target_name);
     vector<memory_info *>::iterator it;
     // cerr << exe_file << " " << hash <<  endl;
     // iterate over the list of memory locations
@@ -384,7 +331,6 @@ bool SHMProcess::Private::validate(char * exe_file, uint32_t pid, vector <memory
             {
                 memory_info * m = *it;
                 memdescriptor = m;
-                process_ID = pid;
                 identified = true;
                 // cerr << "identified " << m->getVersion() << endl;
                 return true;
@@ -427,9 +373,15 @@ int SHMProcess::getPID()
     return d->process_ID;
 }
 
-//FIXME: implement
+// there is only one we care about.
 bool SHMProcess::getThreadIDs(vector<uint32_t> & threads )
 {
+    if(d->attached)
+    {
+        threads.clear();
+        threads.push_back(d->process_ID);
+        return true;
+    }
     return false;
 }
 
@@ -466,7 +418,7 @@ bool SHMProcess::suspend()
     {
         return false;
     }
-    if(d->suspended)
+    if(d->locked)
     {
         return true;
     }
@@ -496,7 +448,6 @@ bool SHMProcess::suspend()
     // we wait for the server to give up our suspend lock (held by default)
     if(lockf(d->suspend_lock,F_LOCK,0) == 0)
     {
-        d->suspended = true;
         d->locked = true;
         return true;
     }
@@ -510,18 +461,17 @@ bool SHMProcess::asyncSuspend()
     {
         return false;
     }
-    if(d->suspended)
+    if(d->locked)
     {
         return true;
     }
-    
-    if(D_SHMCMD == CORE_SUSPENDED)
+    uint32_t cmd = D_SHMCMD;
+    if(cmd == CORE_SUSPENDED)
     {
         // we have to hold the lock to be really suspended
         if(lockf(d->suspend_lock,F_LOCK,0) == 0)
         {
             d->locked = true;
-            d->suspended = true;
             return true;
         }
         return false;
@@ -529,7 +479,11 @@ bool SHMProcess::asyncSuspend()
     else
     {
         // did we just resume a moment ago?
-        if(D_SHMCMD == CORE_RUN)
+        if(cmd == CORE_STEP)
+        {
+            return false;
+        }
+        else if(cmd == CORE_RUN)
         {
             D_SHMCMD = CORE_STEP;
         }
@@ -537,14 +491,6 @@ bool SHMProcess::asyncSuspend()
         {
             D_SHMCMD = CORE_SUSPEND;
         }
-        // try locking
-        if(lockf(d->suspend_lock,F_TLOCK,0) == 0)
-        {
-            d->locked = true;
-            d->suspended = true;
-            return true;
-        }
-        
         return false;
     }
 }
@@ -559,12 +505,11 @@ bool SHMProcess::resume()
 {
     if(!d->attached)
         return false;
-    if(!d->suspended)
+    if(!d->locked)
         return true;
     // unlock the suspend lock
     if(lockf(d->suspend_lock,F_ULOCK,0) == 0)
     {
-        d->suspended = false;
         d->locked = false;
         if(d->SetAndWait(CORE_RUN)) // we have to make sure the server responds!
         {
@@ -587,17 +532,27 @@ bool SHMProcess::attach()
     }
     if(!d->GetLocks())
     {
-        cerr << "server is full or not really there!" << endl;
+        //cerr << "server is full or not really there!" << endl;
         return false;
     }
 
+    /*
+    * Locate the segment.
+    */
+    if ((d->shm_ID = shmget(SHM_KEY + d->process_ID, SHM_SIZE, 0666)) < 0)
+    {
+        d->FreeLocks();
+        cerr << "can't find segment" << endl; // FIXME: throw
+        return false;
+    }
+    
     /*
     * Attach the segment
     */
     if ((d->shm_addr = (char *) shmat(d->shm_ID, NULL, 0)) == (char *) -1)
     {
         d->FreeLocks();
-        cerr << "can't attach segment" << endl;
+        cerr << "can't attach segment" << endl; // FIXME: throw
         return false;
     }
     d->attached = true;
@@ -618,18 +573,18 @@ bool SHMProcess::detach()
     {
         return false;
     }
-    if(d->suspended)
+    if(d->locked)
     {
         resume();
     }
     // detach segment
     if(shmdt(d->shm_addr) != -1)
     {
+        d->FreeLocks();
+        d->locked = false;
         d->attached = false;
-        d->suspended = false;
         d->shm_addr = 0;
         g_pProcess = 0;
-        d->FreeLocks();
         return true;
     }
     // fail if we can't detach
@@ -851,7 +806,6 @@ const std::string SHMProcess::readSTLString(uint32_t offset)
     D_SHMHDR->address = offset;
     full_barrier
     d->SetAndWait(CORE_READ_STL_STRING);
-    //int length = ((shm_retval *)d->my_shm)->value;
     return(string( D_SHMDATA(char) ));
 }
 
@@ -920,4 +874,23 @@ char * SHMProcess::getSHMStart (void)
     if(!d->locked) return 0; //THROW HERE!
         
     return /*d->shm_addr_with_cl_idx*/ d->shm_addr;
+}
+
+bool SHMProcess::Private::Aux_Core_Attach(bool & versionOK, pid_t & PID)
+{
+    if(!locked) throw Error::SHMAccessDenied();
+    
+    SHMDATA(coreattach)->cl_affinity = OS_getAffinity();
+    if(!SetAndWait(CORE_ATTACH)) return false;
+    /*
+    cerr <<"CORE_VERSION" << CORE_VERSION << endl;
+    cerr <<"server CORE_VERSION" << SHMDATA(coreattach)->sv_version << endl;
+    */
+    versionOK =( SHMDATA(coreattach)->sv_version == CORE_VERSION );
+    PID = SHMDATA(coreattach)->sv_PID;
+    useYield = SHMDATA(coreattach)->sv_useYield;
+    #ifdef DEBUG
+        if(useYield) cerr << "Using Yield!" << endl;
+    #endif
+    return true;
 }

shmserver/mod-core.cpp

@@ -183,6 +183,12 @@ void ReleaseSuspendLock( void * data )
     OS_releaseSuspendLock(currentClient);
 }
 
+void AcquireSuspendLock( void * data )
+{
+    OS_lockSuspendLock(currentClient);
+}
+
+
 DFPP_module InitCore(void)
 {
     DFPP_module core;
@@ -193,7 +199,8 @@ DFPP_module InitCore(void)
     core.reserve(NUM_CORE_CMDS);
     // basic states
     core.set_command(CORE_RUNNING, CANCELLATION, "Running");
-    core.set_command(CORE_RUN, FUNCTION, "Run!",0,CORE_RUNNING);
+    //core.set_command(CORE_RUN, FUNCTION, "Run!",AcquireSuspendLock,CORE_RUNNING);
+    core.set_command(CORE_RUN, CANCELLATION, "Run!",0,CORE_RUNNING);
     core.set_command(CORE_STEP, CANCELLATION, "Suspend on next step",0,CORE_SUSPEND);// set command to CORE_SUSPEND, check next client
     core.set_command(CORE_SUSPEND, FUNCTION, "Suspend", ReleaseSuspendLock , CORE_SUSPENDED);
     core.set_command(CORE_SUSPENDED, CLIENT_WAIT, "Suspended");
@@ -300,16 +307,17 @@ void SHM_Act (void)
         if(cmd.nextState != -1)
         {
             /*
-            fprintf(stderr, "Client %d invoked %d:%d = %x = ",
-                    currentClient,((shm_cmd)atomic).parts.module,((shm_cmd)atomic).parts.command, cmd._function);
-            fprintf(stderr, "%s\n",cmd.name.c_str());
+            char text [512];
+            char text2 [512];
+            sprintf (text,"Client %d invoked %d:%d = %x = %s\n",currentClient,((shm_cmd)atomic).parts.module,((shm_cmd)atomic).parts.command, cmd._function,cmd.name.c_str());
+            sprintf(text2, "Server set %d\n",cmd.nextState);
             */
             // FIXME: WHAT HAPPENS WHEN A 'NEXTSTATE' IS FROM A DIFFERENT MODULE THAN 'CORE'? Yeah. It doesn't work.
             SHMCMD = cmd.nextState;
-            /*
-            fprintf(stderr, "Server set %d\n",cmd.nextState);
-            fflush(stderr); // make sure this finds its way to the terminal!
-            */
+            //MessageBox(0,text,text2, MB_OK);
+            
+            //fflush(stderr); // make sure this finds its way to the terminal!
+            
         }
         full_barrier
         

shmserver/shms-windows.cpp

@@ -44,8 +44,65 @@ char *shm = 0;
 int shmid = 0;
 bool inited = 0;
 HANDLE shmHandle = 0;
+
 HANDLE DFSVMutex = 0;
-HANDLE DFCLMutex = 0;
+HANDLE DFCLMutex[SHM_MAX_CLIENTS];
+HANDLE DFCLSuspendMutex[SHM_MAX_CLIENTS];
+int held_DFCLSuspendMutex[SHM_MAX_CLIENTS];
+int numheld = SHM_MAX_CLIENTS;
+
+
+void OS_lockSuspendLock(int which)
+{
+    if(numheld == SHM_MAX_CLIENTS)
+        return;
+    // lock not held by server and can be picked up. OK.
+    if(held_DFCLSuspendMutex[which] == 0)
+    {
+        uint32_t state = WaitForSingleObject(DFCLSuspendMutex[which],INFINITE);
+        if(state == WAIT_ABANDONED || state == WAIT_OBJECT_0)
+        {
+            held_DFCLSuspendMutex[which] = 1;
+            numheld++;
+            return;
+        }
+        // lock couldn't be picked up!
+        errorstate = 1;
+        MessageBox(0,"Suspend lock locking failed. Further communication disabled!","Error", MB_OK);
+        return;
+    }
+    errorstate = 1;
+    MessageBox(0,"Server tried to lock already locked suspend lock?  Further communication disabled!","Error", MB_OK);
+    return;
+}
+
+void OS_releaseSuspendLock(int which)
+{
+    /*
+    if(which >=0 && which < SHM_MAX_CLIENTS)
+        return;
+    */
+    if(numheld != SHM_MAX_CLIENTS)
+    {
+        MessageBox(0,"Locking system failure. Further communication disabled!","Error", MB_OK);
+        errorstate = 1;
+        return;
+    }
+    // lock hel by server and can be released -> OK
+    if(held_DFCLSuspendMutex[which] == 1 && ReleaseMutex(DFCLSuspendMutex[which]))
+    {
+        numheld--;
+        held_DFCLSuspendMutex[which] = 0;
+    }
+    // locked and not can't be released? FAIL!
+    else if (held_DFCLSuspendMutex[which] == 1)
+    {
+        MessageBox(0,"Suspend lock failed to unlock.  Further communication disabled!","Error", MB_OK);
+        return;
+    }
+}
+
+
 void SHM_Init ( void )
 {
     // check that we do this only once per process
@@ -56,102 +113,81 @@ void SHM_Init ( void )
     }
     inited = true;
     
-    char svmutexname [256];
-    sprintf(svmutexname,"DFSVMutex-%d",OS_getPID());
+    
     char clmutexname [256];
-    sprintf(clmutexname,"DFCLMutex-%d",OS_getPID());
+    char clsmutexname [256];
     char shmname [256];
     sprintf(shmname,"DFShm-%d",OS_getPID());    
     
-    // create or open mutexes
+    // create a locked server mutex
+    char svmutexname [256];
+    sprintf(svmutexname,"DFSVMutex-%d",OS_getPID());
     DFSVMutex = CreateMutex( 0, 1, svmutexname);
     if(DFSVMutex == 0)
     {
-        DFSVMutex = OpenMutex(SYNCHRONIZE,false, svmutexname);
-        if(DFSVMutex == 0)
-        {
-            errorstate = 1;
-            return;
-        }
+        MessageBox(0,"Server mutex creation failed. Further communication disabled!","Error", MB_OK);
+        errorstate = 1;
+        return;
     }
-    DFCLMutex = CreateMutex( 0, 0, clmutexname);
-    if(DFCLMutex == 0)
+    // the mutex already existed. we don't want to know.
+    if(GetLastError() == ERROR_ALREADY_EXISTS)
     {
-        DFCLMutex = OpenMutex(SYNCHRONIZE,false, clmutexname);
-        if(DFCLMutex == 0)
-        {
-            CloseHandle(DFSVMutex);
-            errorstate = 1;
-            return;
-        }
+        MessageBox(0,"Server mutex already existed. Further communication disabled!","Error", MB_OK);
+        errorstate = 1;
+        return;
     }
     
-    // try locking server mutex
-    uint32_t result;
-    result = WaitForSingleObject(DFSVMutex,0);
-    switch (result)
+    // create client and suspend mutexes
+    for(int i = 0; i < SHM_MAX_CLIENTS; i++)
     {
-        case WAIT_ABANDONED:
+        sprintf(clmutexname,"DFCLMutex-%d-%d",OS_getPID(),i);
+        sprintf(clsmutexname,"DFCLSuspendMutex-%d-%d",OS_getPID(),i);
+        
+        DFCLMutex[i] = CreateMutex( 0, 0, clmutexname); // client mutex, not held
+        DFCLSuspendMutex[i] = CreateMutex( 0, 1, clsmutexname); // suspend mutexes held on start
+        held_DFCLSuspendMutex[i] = 1;
+        
+        if(DFCLMutex[i] == 0 || DFCLSuspendMutex[i] == 0 || GetLastError() == ERROR_ALREADY_EXISTS)
         {
-            // picked up after a crashed DF process
-            // do some sanity checks on client attached
-            // otherwise the same thing as WAIT_OBJECT_0
-            break;
-        }
-        case WAIT_OBJECT_0:
-        {
-            // all right, we have the mutex and are the one and only DF server
-            break;
-        }
-        case WAIT_TIMEOUT:
-        case WAIT_FAILED:
-        default:
-        {
-            // error, bail
+            MessageBox(0,"Client mutex creation failed. Close all tools before starting DF.","Error", MB_OK);
             errorstate = 1;
-            MessageBox(0,"Could not aquire mutex","Error", MB_OK);
-            CloseHandle(DFSVMutex);
-            CloseHandle(DFCLMutex);
             return;
         }
     }
-    
+
     // create virtual memory mapping
-    shmHandle = CreateFileMapping(INVALID_HANDLE_VALUE,NULL,PAGE_READWRITE,0,SHM_ALL_CLIENTS,shmname);
+    shmHandle = CreateFileMapping(INVALID_HANDLE_VALUE,NULL,PAGE_READWRITE,0,SHM_SIZE,shmname);
     // if can't create or already exists -> nothing happens
     if(GetLastError() == ERROR_ALREADY_EXISTS)
     {
         MessageBox(0,"SHM bridge already in use","Error", MB_OK);
         errorstate = 1;
-        ReleaseMutex(DFSVMutex);
-        CloseHandle(DFSVMutex);
-        CloseHandle(DFCLMutex);
         return;
     }
     if(!shmHandle)
     {
         MessageBox(0,"Couldn't create SHM bridge","Error", MB_OK);
         errorstate = 1;
-        ReleaseMutex(DFSVMutex);
-        CloseHandle(DFSVMutex);
-        CloseHandle(DFCLMutex);
         return;
     }
     // attempt to attach the created mapping
-    shm = (char *) MapViewOfFile(shmHandle,FILE_MAP_ALL_ACCESS, 0,0, SHM_ALL_CLIENTS);
+    shm = (char *) MapViewOfFile(shmHandle,FILE_MAP_ALL_ACCESS, 0,0, SHM_SIZE);
     if(shm)
     {
-        ((shm_cmd *)shm)->pingpong = CORE_RUNNING;
+        // make sure we don't stall or do crazy stuff
+        for(int i = 0; i < SHM_MAX_CLIENTS;i++)
+        {
+            ((uint32_t *)shm)[i] = CORE_RUNNING;
+        }
+        // init modules :)
+        InitModules();
     }
     else
     {
         MessageBox(0,"Couldn't attach SHM bridge","Error", MB_OK);
         errorstate = 1;
-        ReleaseMutex(DFSVMutex);
-        CloseHandle(DFSVMutex);
-        CloseHandle(DFCLMutex);
+        return;
     }
-    InitModules();
 }
 
 void SHM_Destroy ( void )
@@ -159,9 +195,13 @@ void SHM_Destroy ( void )
     if(errorstate)
         return;
     KillModules();
-    ReleaseMutex(DFSVMutex);
+    // get rid of all the locks
     CloseHandle(DFSVMutex);
-    CloseHandle(DFCLMutex);
+    for(int i=0; i < SHM_MAX_CLIENTS; i++)
+    {
+        CloseHandle(DFCLSuspendMutex[i]);
+        CloseHandle(DFCLMutex[i]);
+    }
 }
 
 uint32_t OS_getPID()
@@ -181,17 +221,18 @@ uint32_t OS_getAffinity()
 
 
 // is the other side still there?
-bool isValidSHM()
+bool isValidSHM(int which)
 {
-    // try if CL mutex is free
-    uint32_t result = WaitForSingleObject(DFCLMutex,0);
+    // try if CL mutex is free (by locking client mutex)
+    uint32_t result = WaitForSingleObject(DFCLMutex[which],0);
     
     switch (result)
     {
         case WAIT_ABANDONED:
         case WAIT_OBJECT_0:
         {
-            ReleaseMutex(DFCLMutex);
+            OS_lockSuspendLock(which);
+            ReleaseMutex(DFCLMutex[which]);
             return false;
         }
         case WAIT_TIMEOUT: