PHP concurrency issue, multiple simultaneous requests; mutexes?

Question

So I've just realised that PHP is potentially running multiple requests simultaneously. The logs from last night seem to show that two requests came in, were processed in parallel; each triggered an import of data from another server; each attempted to insert a record into the database. One request failed when it tried to insert a record that the other thread had just inserted (the imported data comes with PKs; I'm not using incrementing IDs): SQLSTATE[23000]: Integrity constraint violation: 1062 Duplicate entry '865020' for key 'PRIMARY' ....

1. Have I diagnosed this issue correctly?
2. How should I address this?

The following is some of the code. I've stripped out much of it (the logging, the creation of other entities beyond the Patient from the data), but the following should include the relevant snippets. Requests hit the import() method, which calls importOne() for each record to import, essentially. Note the save method in importOne(); that's an Eloquent method (using Laravel and Eloquent) that will generate the SQL to insert/update the record as appropriate.

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();
}

I'd guess that the solution would require a mutex around the entire import block? And if a request couldn't attain a mutex, it'd simply move on with the rest of the request. Thoughts?

EDIT: Just to note, this isn't a critical failure. The exception is caught and logged, and then the request is responded to as per usual. And the import succeeds on the other request, and then that request is responded to as per usual. The users are none-the-wiser; they don't even know about the import, and that isn't the main focus of the request coming in. So really, I could just leave this running as is, and aside from the occasional exception, nothing bad happens. But if there is a fix to prevent additional work being done/multiple requests being sent of to this other server unnecessarily, that could be worth pursuing.

EDIT2: Okay, I've taken a swing at implementing a locking mechanism with flock(). Thoughts? Would the following work? And how would I unit test this addition?

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: Thoughts on the following alternate implementation of the lock:

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

Your example code would block the second request until the first is finished. You would need to use LOCK_NB option for flock() to return error immediately and not wait.

Yes you can use either locking or semaphores, either on filesystem level or directly in the database.

In your case when you need each import file to be processed only once, the best solution would be to have a SQL table with row for each import file. At the beginning of import, you insert the info that import is in progress, so other threads will know to not process it again. After import is finished, you mark it as such. (Then few hours later you can check the table to see if the import really finished.)

Also it is better to do such one-time long-lasting things like import on separate scripts and not while serving normal webpages to visitors. For example you can schedule a nightly cron job which would pick up the import file and process it.

Answer 2

It doesn't seem like you are having a race condition, because the ID is coming from the import file, and if your import algorithm is working correctly then each thread would have its own shard of the work to be done and should never conflict with others. Now it seems like 2 threads are receiving a request to create the same patient and get in conflict with eachother because of bad algorithm.

Make sure that each spawned thread gets a new row from the import file, and repeat only on failure.

If you cant do that, and want to stick to mutex, using a file lock doesn't seem like a very nice solution, since now you solved the conflict within the application, while it is actually occurring in your database. A DB lock should be a lot faster too, and overall a more decent solution.

Request a database lock, like this:

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

And you can expect a SQL error when you would write to a locked table, so surround your Patient->save() with a try catch.

An even better solution would be to use a conditional atomic query. A DB query that also has the condition within it. You could use a query like this:

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

Answer 3

I see three options:
- use mutex/semaphore/some other flag - not easy to code and maintain
- use DB built-in transaction mechanism
- use queue (like RabbitMQ or 0MQ) to write messages into DB in a row