Cross Process Event - Rilasciare tutti i camerieri in modo affidabile

Question

ho creato un evento di processo croce via ManualResetEvent. Quando si verifica questo evento potenzialmente n thread in n processi diversi dovrebbero essere sbloccati e iniziare a funzionare per recuperare i nuovi dati. Il problema è che sembra che ManualResetEvent.Set seguito da un Reset immediato non causi il risveglio di tutti i thread in attesa. La documentazione è piuttosto vago ci

http://msdn.microsoft.com/en-us/library/windows/desktop/ms682396(v=vs.85).aspx

Quando lo stato di un oggetto evento manuale di reset viene segnalata, rimane segnalato finché non viene esplicitamente riportato a non segnalato dalla funzione ResetEvent . Qualsiasi numero di thread in attesa, o fili che successivamente iniziano operazioni di attesa per l'oggetto evento specificato, può essere rilasciato mentre lo stato dell'oggetto viene segnalata.

C'è un metodo chiamato PulseEvent che sembra fare esattamente ciò di cui ho bisogno ma sfortunatamente è anche difettoso.

Un thread in attesa su un oggetto di sincronizzazione può essere momentaneamente rimosso dallo stato di attesa da una modalità kernel APC, ed allora rinviato stato di attesa dopo l'APC è completa. Se la chiamata a PulseEvent si verifica durante il periodo in cui il filo è stato rimosso dallo stato di attesa , il filo non sarà rilasciato perché PulseEvent stampa solo quei thread nell'attesa al momento in cui viene chiamato. Pertanto, PulseEvent è inaffidabile e non deve essere utilizzato dai nuovi applicazioni. Invece, usa le variabili di condizione.

Ora MS consiglia di utilizzare le variabili di condizione.

variabili di condizione sono primitive di sincronizzazione che consentono discussioni attendere finché non si verifica una condizione particolare. variabili di condizione sono oggetti modalità utente che non possono essere condivise tra i processi.

In seguito ai documenti mi sembra di aver perso la fortuna di farlo in modo affidabile. C'è un modo semplice per realizzare la stessa cosa senza le limitazioni indicate con un solo ManualResetEvent o devo creare per ogni ascoltatore un evento di risposta per ottenere un ACK per ogni chiamante iscritto? In tal caso avrei bisogno di una piccola memoria condivisa per registrare i pid dei processi sottoscritti ma sembra che porti una serie di problemi. Cosa succede quando un processo si blocca o non risponde? ....

di dare qualche contesto. Ho un nuovo stato da pubblicare che tutti gli altri processi dovrebbero leggere da una posizione di memoria condivisa. Va bene perdere un aggiornamento quando si verificano più aggiornamenti contemporaneamente ma il processo deve leggere almeno l'ultimo valore aggiornato. Potrei sondare con un timeout ma non sembra una soluzione corretta.

Attualmente sto giù a

ChangeEvent = new EventWaitHandle(false, EventResetMode.ManualReset, counterName + "_Event"); 

ChangeEvent.Set(); 
Thread.Sleep(1); // increase odds to release all waiters 
ChangeEvent.Reset(); 

Answer 1

Un'opzione generica per gestire il caso in cui i produttori devono svegliare tutti i consumatori e il numero di consumatori si sta evolvendo consiste nell'utilizzare un approccio di recinzione mobile. Questa opzione richiede anche una regione IPC con memoria condivisa. Il metodo a volte porta al risveglio dei consumatori quando non è presente alcun lavoro, soprattutto se molti processi richiedono una pianificazione e il carico è elevato, ma si sveglieranno sempre eccetto che su macchine irrimediabilmente sovraccariche.

Creare diversi eventi di ripristino manuale e chiedere ai produttori di mantenere un contatore per il prossimo evento che verrà impostato. Tutti gli eventi vengono lasciati impostati, tranne l'evento NextToFire. I processi di consumo attendono l'evento NextToFire. Quando il produttore desidera riattivare tutti i consumatori, reimposta l'evento Next + 1 e imposta l'evento corrente. Tutti i consumatori saranno programmati e quindi attenderanno il nuovo evento NextToFire. L'effetto è che solo il produttore utilizza ResetEvent, ma i consumatori sanno sempre quale evento sarà il prossimo a svegliarli.

Tutti gli utenti Init: (pseudo codice è C/C++, non C#)

// Create Shared Memory and initialise NextToFire; 
pSharedMemory = MapMySharedMemory(); 
if (First to create memory) pSharedMemory->NextToFire = 0; 

HANDLE Array[4]; 
Array[0] = CreateEvent(NULL, 1, 0, "Event1"); 
Array[1] = CreateEvent(NULL, 1, 0, "Event2"); 
Array[2] = CreateEvent(NULL, 1, 0, "Event3"); 
Array[3] = CreateEvent(NULL, 1, 0, "Event4"); 

Produttore Wake tutti

long CurrentNdx = pSharedMemory->NextToFire; 
long NextNdx = (CurrentNdx+1) & 3; 

// Reset next event so consumers block 
ResetEvent(Array[NextNdx]); 

// Flag to consumers new value 
long Actual = InterlockedIncrement(&pSharedMemory->NextToFire) & 3; 

// Next line needed if multiple producers active. 
// Not a perfect solution 
if (Actual != NextNdx) ResetEvent(Actual); 

// Now wake them all up 
SetEvent(CurrentNdx); 

I consumatori aspettano logica

long CurrentNdx = (pSharedMemory->NextToFire) & 3; 
WaitForSingleObject(Array[CurrentNdx], Timeout); 

Answer 2

Dal .NET 4.0, è possibile utilizzare file mappato in memoria per la sincronizzazione memoria di processo. In questo caso, scrivere contatore su MemoryMappedFile e decrementarlo dai processi di lavoro. Se il contatore è uguale a zero, allora il processo principale ha permesso di resettare l'evento. Ecco il codice di esempio.

processo principale

//number of WorkerProcess 
int numWorkerProcess = 5; 

//Create MemroyMappedFile object and accessor. 4 means int size. 
MemoryMappedFile mmf = MemoryMappedFile.CreateNew("test_mmf", 4); 
MemoryMappedViewAccessor accessor = mmf.CreateViewAccessor(); 

EventWaitHandle ChangeEvent = new EventWaitHandle(false, EventResetMode.ManualReset, counterName + "_Event"); 

//write counter to MemoryMappedFile 
accessor.Write(0, numWorkerProcess); 

//..... 

ChangeEvent.Set(); 

//spin wait until all workerProcesses decreament counter 
SpinWait.SpinUntil(() => { 

    int numLeft = accessor.ReadInt32(0); 
    return (numLeft == 0); 
}); 


ChangeEvent.Reset(); 

WorkerProcess

//Create existed MemoryMappedfile object which created by main process. 
MemoryMappedFile mmf = MemoryMappedFile.OpenExisting("test_mmf"); 
MemoryMappedViewAccessor accessor = mmf.CreateViewAccessor(); 

//This mutex object is used for decreament counter. 
Mutex mutex = new Mutex(false, "test_mutex"); 
EventWaitHandle ChangeEvent = new EventWaitHandle(false, EventResetMode.ManualReset, "start_Event"); 

//.... 

ChangeEvent.WaitOne(); 

//some job... 

//decrement counter with mutex lock. 
mutex.WaitOne(); 
int count = accessor.ReadInt32(0); 
--count; 
accessor.Write(0, count); 
mutex.ReleaseMutex(); 
///////////////////////////////////// 

Se l'ambiente è inferiore a .NET 4.0, si potrebbe realizzare utilizzando la funzione CreateFileMapping da win32 API.

Answer 3

Lei ha scritto: "PulseEvent che sembra fare esattamente ciò di cui ho bisogno ma sfortunatamente è anche difettoso". Questo è vero che PulseEvent è difettoso, ma non posso essere d'accordo sul fatto che l'evento di reset manuale sia difettoso. È molto affidabile Esistono solo casi in cui è possibile utilizzare eventi di ripristino manuale e in alcuni casi non è possibile utilizzarli. Non è adatto a tutti. Esistono molti altri strumenti, come eventi di reset automatico, pipe, ecc.

Il modo migliore per notificare un thread, se è necessario notificarlo periodicamente, ma non è necessario inviare i dati tra i processi, è un evento di reset automatico. Hai solo bisogno del proprio evento per ogni discussione. Quindi, hai tanti eventi quanti sono i thread.

Se è necessario inviare semplicemente i dati ai processi, è preferibile utilizzare named pipe. A differenza degli eventi di reimpostazione automatica, non è necessaria la propria pipe per ciascuno dei processi. Ogni pipe denominata ha un server e uno o più client. Quando ci sono molti client, molte istanze delle stesse named pipe vengono create automaticamente dal sistema operativo per ciascuno dei client. Tutte le istanze di una pipe denominata condividono lo stesso nome pipe, ma ciascuna istanza ha i propri buffer e handle e fornisce un conduit separato per le comunicazioni client/server. L'utilizzo delle istanze consente a più client pipe di utilizzare contemporaneamente la stessa named pipe. Qualsiasi processo può agire sia come server per una pipe che come client per un'altra pipe, e viceversa, rendendo possibile la comunicazione peer-to-peer.

Se si utilizzerà una named pipe, non ci sarà alcun bisogno degli eventi nel proprio scenario, e i dati avranno una consegna garantita indipendentemente da ciò che accade con i processi - ognuno dei processi potrebbe ottenere lunghi ritardi (ad esempio con uno swap) ma i dati saranno finalmente consegnati al più presto senza il tuo coinvolgimento speciale.

Un evento per tutti i thread (processi) è OK solo se la notifica sarà una sola volta. In questo caso, avrai bisogno dell'evento di reset manuale, non di un reset automatico. Ad esempio, se hai bisogno di notificare che la tua applicazione uscirà molto presto, potresti segnalare questo evento di reset manuale comune. Ma, come ho scritto prima, nel tuo scenario, named pipe è la scelta migliore.