Coerenza dei dati nei sistemi distribuiti

Utilizzando una struttura di (1) bilanciamento del carico, (2) lavoratori e (3) un raccoglitore di risultati:

Il bilanciamento del carico assegna un numero identificativo incrementale a ciascun elemento in entrata, quindi trasmette la combinazione del nuovo elemento con il suo numero identificativo a tutti i lavoratori.

I lavoratori identificano potenziali corrispondenze e inviano candidati accoppiati al raccoglitore.

Il gatherer riceve tutti i candidati delle coppie di match dagli operai e ha una funzione di accettazione, come minimamente, scegliendo la prima coppia in cui entrambi gli elementi non sono ancora stati abbinati. Dopo l'accettazione di una coppia, il raccoglitore trasmette ulteriormente, di nuovo agli operai, i singoli elementi delle coppie accettate in modo che possano smettere di lavorare su quegli elementi.

Al centro dell'algoritmo del lavoratore è che accettano di suddividere il problema in anticipo.

Ogni worker è configurato con due costanti integer: un numero univoco di worker e il numero totale di worker. I lavoratori sono programmati per utilizzare tali costanti per suddividere il lavoro in modo che ciascuno lavori su parti diverse dello spazio di ricerca di potenziali corrispondenze.

I lavoratori ricevono (1) nuovi elementi dell'elemento di corrispondenza (numerati) dal servizio di bilanciamento del carico e (2) elementi ritirati dal raccoglitore.

Ad esempio, un lavoratore testa gli elementi per le partite come segue, dato 2 lavoratori totali:

• worker 1 verifica le corrispondenze, quando gli elementi provengono dal servizio di bilanciamento del carico:

  • elemento 1 con elemento 2
  • elemento 1 con elemento 3
  • elemento 1 con elemento 4 ...
  • elemento 3 con elemento 4
  • elemento 3 con elemento 5 ...
  • elemento 5 con elemento 6 ...

• worker 2 test per le corrispondenze, quando gli elementi arrivano:

  • elemento 2 con elemento 3
  • elemento 2 con elemento 4 ...
  • elemento 4 con elemento 5
  • elemento 4 con elemento 6 ...
  • elemento 6 con elemento 7 ...

Quando l'elemento 1 è noto come lavoratore eliminato 1 smette di trovare i candidati per l'elemento 1, ecc ...

(Esistono anche potenziali ottimizzazioni che richiederebbero essenzialmente un maggiore coordinamento.)

Il bilanciamento del carico può essere facilmente ridimensionato suddividendo lo spazio numerico del contatore incrementale (ad es. evens / odds come una suddivisione a due vie).

Se stai implementando la coda in un database, puoi ottenere l'esclusione dell'aggiornamento reciproco utilizzando la transazione REPEATABLE READ livello di isolamento . La tabella consentirà comunque gli inserimenti, ma tutte le righe che vengono lette dal tuo codice rimarranno non modificabili fino al termine della transazione.

Quindi se un lavoratore vuole afferrare i prossimi due elementi disponibili, può prenderli con un semplice SQL.

Supponiamo che tu abbia una tabella chiamata Tasks con una chiave primaria di ID e una colonna denominata WorkerID , che è un identificatore che indica quale lavoratore può lavorare sull'attività. WorkerID è impostato su NULL per iniziare. Per afferrare i prossimi due elementi si dovrebbe eseguire il seguente comando:

UPDATE TOP (2) Tasks
SET WorkerID = @MyWorkerID
WHERE WorkerID IS NULL

Naturalmente, se il database è replicato, questo tipo di schema funzionerebbe solo con la replica transazionale .

Sembra qualcosa che ho fatto in passato. Suppongo che la coda si trovi su un'istanza di database.

loop forever
  begin transaction
  var firstMatch = null
  foreach row order by sequentialId
    update a column in row
    if write succeeds then
      firstMatch = sequentialId
      break foreach
    end if
  end foreach
  if firstMatch != null then
    foreach row where sequentialId > firstMatch order by sequentialId
      if row matches criteria then
        update a column in the row
        if update succeeds then
          delete firstMatch
          delete secondMatch
          break foreach
        end if
      end if
    end foreach
  end if
  commit transaction
end loop

Puoi avere quanti processi / thread vuoi usare questo codice. In pratica stai mettendo i blocchi di scrittura sui record che ti interessano e un'altra transazione non sarà in grado di afferrare quel record. Inoltre, poiché stai ordinando il tuo foreach usando sequentialId, non puoi mai ottenere una condizione di deadlock.