Problema di concorrenza PHP, più richieste simultanee; mutex?

Question

Quindi mi sono appena reso conto che PHP è potenzialmente in grado di eseguire più richieste contemporaneamente. I log della scorsa notte sembrano mostrare che sono arrivate due richieste, sono state elaborate in parallelo; ognuno ha attivato un'importazione di dati da un altro server; ognuno ha tentato di inserire un record nel database. Una richiesta non è riuscita quando ha tentato di inserire un record che l'altro thread aveva appena inserito (i dati importati vengono forniti con PK, non sto utilizzando ID incrementali): SQLSTATE[23000]: Integrity constraint violation: 1062 Duplicate entry '865020' for key 'PRIMARY' ....

1. Ho diagnosticato correttamente questo problema?
2. Come devo affrontare questo?

Quanto segue è parte del codice. Ne ho tolto gran parte (la registrazione, la creazione di altre entità oltre il Paziente dai dati), ma i seguenti dovrebbero includere i frammenti rilevanti. Le richieste colpiscono il metodo import(), che chiama importOne() per ogni record da importare, essenzialmente. Nota il metodo di salvataggio in importOne(); questo è un metodo Eloquent (usando Laravel ed Eloquent) che genererà l'SQL per inserire/aggiornare il record come appropriato.

public function import() 
{ 
     $now = Carbon::now(); 
     // Get data from the other server in the time range from last import to current import 
     $calls = $this->getCalls($this->getLastImport(), $now); 
     // For each call to import, insert it into the DB (or update if it already exists) 
     foreach ($calls as $call) { 
      $this->importOne($call); 
     } 
     // Update the last import time to now so that the next import uses the correct range 
     $this->setLastImport($now); 
} 

private function importOne($call) 
{ 
    // Get the existing patient for the call, or create a new one 
    $patient = Patient::where('id', '=', $call['PatientID'])->first(); 
    $isNewPatient = $patient === null; 
    if ($isNewPatient) { 
     $patient = new Patient(array('id' => $call['PatientID'])); 
    } 
    // Set the fields 
    $patient->given_name = $call['PatientGivenName']; 
    $patient->family_name = $call['PatientFamilyName']; 
    // Save; will insert/update appropriately 
    $patient->save(); 
} 

Suppongo che la soluzione richiederebbe un mutex attorno all'intero blocco di importazione? E se una richiesta non potesse raggiungere un mutex, sarebbe semplicemente andare avanti con il resto della richiesta. Pensieri?

EDIT: Solo per notare, questo non è un fallimento critico. L'eccezione viene catturata e registrata e quindi la richiesta viene inviata come al solito. E l'importazione ha successo con l'altra richiesta, e quindi quella richiesta viene data come al solito. Gli utenti sono nessuno-saggi; non sanno nemmeno dell'importazione, e questo non è l'obiettivo principale della richiesta in arrivo. Quindi, davvero, potrei lasciarlo così com'è e, a parte l'eccezione occasionale, non succede niente di male. Ma se c'è una soluzione per evitare che venga fatto del lavoro aggiuntivo/richieste multiple inviate a questo altro server inutilmente, potrebbe valere la pena di proseguire.

EDIT2: Ok, ho implementato un meccanismo di blocco con flock(). Pensieri? Il seguente lavoro sarebbe? E come posso testare questa aggiunta?

public function import() 
{ 
    try { 
     $fp = fopen('/tmp/lock.txt', 'w+'); 
     if (flock($fp, LOCK_EX)) { 
      $now = Carbon::now(); 
      $calls = $this->getCalls($this->getLastImport(), $now); 
      foreach ($calls as $call) { 
       $this->importOne($call); 
      } 
      $this->setLastImport($now); 
      flock($fp, LOCK_UN); 
      // Log success. 
     } else { 
      // Could not acquire file lock. Log this. 
     } 
     fclose($fp); 
    } catch (Exception $ex) { 
     // Log failure. 
    } 
} 

Edit3: Pensieri sulla seguente implementazione alternativa del blocco:

public function import() 
{ 
    try { 
     if ($this->lock()) { 
      $now = Carbon::now(); 
      $calls = $this->getCalls($this->getLastImport(), $now); 
      foreach ($calls as $call) { 
       $this->importOne($call); 
      } 
      $this->setLastImport($now); 
      $this->unlock(); 
      // Log success 
     } else { 
      // Could not acquire DB lock. Log this. 
     } 
    } catch (Exception $ex) { 
     // Log failure 
    } 
} 

/** 
* Get a DB lock, returns true if successful. 
* 
* @return boolean 
*/ 
public function lock() 
{ 
    return DB::SELECT("SELECT GET_LOCK('lock_name', 1) AS result")[0]->result === 1; 
} 

/** 
* Release a DB lock, returns true if successful. 
* 
* @return boolean 
*/ 
public function unlock() 
{ 
    return DB::select("SELECT RELEASE_LOCK('lock_name') AS result")[0]->result === 1; 
} 

Answer 1

vedo tre opzioni:
- uso mutex/semaforo/qualche altra bandiera - non facile da codice e mantenere
- utilizzare il meccanismo di transazione integrato DB
- utilizzare la coda (come RabbitMQ o 0MQ) per scrivere messaggi in DB di fila

Answer 2

Non sembra come o hai una condizione di competizione, perché l'ID proviene dal file di importazione e, se il tuo algoritmo di importazione funziona correttamente, ogni thread avrà il suo frammento del lavoro da svolgere e non dovrebbe mai entrare in conflitto con gli altri. Ora sembra che 2 thread ricevano una richiesta per creare lo stesso paziente e entrare in conflitto tra loro a causa dell'algoritmo sbagliato.

assicurarsi che ogni thread spawn ottiene una nuova riga dal file di importazione, e solo ripetere in caso di fallimento.

Se non si riesce a farlo e si desidera attenersi al mutex, l'utilizzo di un blocco file non sembra una soluzione molto utile, poiché ora è stato risolto il conflitto all'interno dell'applicazione, mentre è effettivamente presente nel database. Un blocco DB dovrebbe essere molto più veloce, e nel complesso una soluzione più decente.

Richiedere un blocco del database, in questo modo:

$ db -> exec ('LOCK TABLES table1 scrivere, table2 WRITE');

E ci si può aspettare un errore SQL quando si scrive su una tabella bloccata, quindi circondare Patient-> save() con un try catch.

Una soluzione ancora migliore sarebbe utilizzare una query atomica condizionale. Una query DB che ha anche la condizione al suo interno. È possibile utilizzare una query come questa:

INSERT INTO targetTable(field1) 
SELECT field1 
FROM myTable 
WHERE NOT(field1 IN (SELECT field1 FROM targetTable)) 

Answer 3

Il codice di esempio bloccherebbe la seconda richiesta fino al termine del primo. È necessario utilizzare l'opzione LOCK_NB per flock() per restituire immediatamente l'errore e non attendere.

Sì, è possibile utilizzare il blocco o i semafori, a livello di file system o direttamente nel database.

Nel caso in cui sia necessario elaborare ciascun file di importazione solo una volta, la soluzione migliore sarebbe disporre di una tabella SQL con riga per ogni file di importazione. All'inizio dell'importazione, si inseriscono le informazioni che l'importazione è in corso, quindi altri thread sapranno di non elaborarlo nuovamente. Al termine dell'importazione, lo contrassegni come tale. (Quindi qualche ora dopo puoi controllare la tabella per vedere se l'importazione è davvero finita.)

Inoltre, è meglio fare cose di una volta di lunga durata come l'importazione su script separati e non mentre si servono le normali pagine web ai visitatori . Ad esempio, è possibile pianificare un processo cron notturno che preleva il file di importazione e lo elabora.