2

I wanted to ask this question that has been bugging my head for a long time. Recently, I have started to develop a distributed system which has continuous and frequent database lookups in a loop. Let me describe it to you.

There are entries in a queue, which can be matched mutually among themselves. For example, let's say we have a queue like :

A B C D E F

With certain rules, these can be matched like, A-B, C-E, D-F . Who matches with who and why is irrelevant for this question I think. One thing that is important is that there can only be ONE match for an element, and then it has to leave the queue.

A program is required to continuously work on this queue (or list, if we ignore the sequence) to find matches, and shrink the queue as fast as possible.

Assuming the number of elements in this queue can be very large, I think there should be multiple programs who work on this queue. So one thing I thought of was creating multiple nodes that run this program, which are called "Matchers" .

The problem is, if matcher1 matches A-B at a certain moment, and Matcher2 matches B-C , we have a race condition for B. Given the distributed nature of the matchers, the synchronization can possibly be on a database that provide consistency guarantee. Like when a is matched, it could be marked on the database which keeps the queue. However there seems to be no reliable way to be sure that other matchers will have the happens-before relationship with this operation, hence no guarantee that the change is observed. Especially if the database used is sharded or distributed so there needs to be some time for propagation. So I am not certain how well this would work.

Another solution I have come up with was, assigning certain groups in the queue exclusively to only one matcher. For example,

Matcher1 has A B C D Matcher2 has E F G H

Now matcher1 only matches A-B-C-D among themselves and matcher2 E-F-G-H. Hence, it is possible to keep Matchers with only one thread, so no race condition occurs. Or we could use a local mutex system to lock upon acknowledging a match, to see if there was another match at that moment, so this way we can also use multithreading in the nodes, while being safe with race conditions.

I am aware, what I have written might lack some coherency, but that exactly reflects how it is in my mind. I am fairly proficient with multithreading and parallelism, however I have never seen a real high end real time system, with race condition issues implemented, so I lack in the experience department.

I wanted to get some feedback on my ideas, and maybe receive some better ideas from you guys. Please direct me to fix my question, in case it is lacking severely.

EDIT: This question has very little to do with methods to synchronise a program running on a single machine. The same program is running on multiple nodes in a cluster, and they have to be synchronised.

Improve this question
8
  • Is the queue on only one machine? In your second solution how do you match D-F since D and F are not in the same matcher?
    – Bogdan
    Dec 26, 2016 at 15:51
  • @Bogdan I did not say I match D-F in the second solution. Actually that is the point. When I break the queue into pieces, elements assigned to different matchers, cannot match together, which is kinda bad. But in a system where millions of queue elements exist, it wouldnt matter too much. And I thought of implementing a timeout for a queue element so if it cannot be matched for some time in its own matcher, it is to be sent to another matcher.
    – Ozum Safa
    Dec 26, 2016 at 16:41
  • @Bogdan To answer your other question, because there will be multiple matchers which are supposed to access the queue, the queue would probably be in a database (like mongo, or redis) Although I am not sure that is the right approach. Maybe ultimately, the queue is persisted in a database, but matchers load the group they are supposed to work on to memory
    – Ozum Safa
    Dec 26, 2016 at 16:41
  • 2
    Have you looked at Apache Kafka?
    – Bogdan
    Dec 26, 2016 at 17:06
  • I keep hearing, though I haven't. I will look into it now.
    – Ozum Safa
    Dec 26, 2016 at 17:34

3 Answers 3

Reset to default
2

Using a structure of (1) load balancer, (2) workers, and (3) a results gatherer:

The load balancer assigns an incrementing id number to each incoming element, then broadcasts the combination of the new element with its id number together to all the workers.

The workers identify potential matches and send matched pair candidates to the gatherer.

The gatherer receives all match pair candidates from the workers and has an acceptance function, such as minimally, choosing the first pair where both elements have not yet been matched. Upon acceptance of a pair, the gatherer further broadcasts, back to the workers, the individual elements of accepted pairs so they can stop working on those elements.

At the core of the worker algorithm is that they agree to subdivide the problem in advance.

Each worker is configured with two integer constants: a unique worker number, and the total number of workers. The workers are programmed to use those constants to subdivide the work so they each work on different parts of the search space of potential matches.

Workers receive (1) new match element candidates (numbered) from the load balancer, and (2) retired elements from the gatherer.

As an example, a worker tests elements for matches as follows, given 2 total workers:

  • worker 1 tests for matches, as elements come in from the load balancer:

    • element 1 with element 2
    • element 1 with element 3
    • element 1 with element 4 ...
    • element 3 with element 4
    • element 3 with element 5 ...
    • element 5 with element 6 ...

  • worker 2 tests for matches, as elements come in:

    • element 2 with element 3
    • element 2 with element 4 ...
    • element 4 with element 5
    • element 4 with element 6 ...
    • element 6 with element 7 ...

When element 1 is known as eliminated worker 1 stops finding match candidates for element 1, etc...

(There are also potential optimizations that would essentially require more coordination.)

The load balancer can easily be scaled by subdividing the number space of the incrementing counter (e.g. evens/odds as a two-way split).

Improve this answer
10
  • Thank you for your answer. This is very similar to what I have done until now. Just as you have said, I have multiple workers and one gatherer. In my case though, I name the gatherer "queuer" and it is also responsible as the load balancer, because it assigns the workers the jobs to work on. Which means matchers only work once, per given job piece and inform the result back to queuer. In both these solutions, it only works with ONE GATHERER. I was more puzzled about that, whether it could create a bottleneck or not. Because using more than one, creates a race condition.
    – Ozum Safa
    Dec 27, 2016 at 12:10
  • Although, in your solution, the entity i named queuer gets rid of the responsibility to promote queues, hence stops acting as a load balancer, which seriously reduces the work load, and the possibility of bottleneck is reduced. So, I like yours better. What existing solutions would you use to persist these queue elements ? Would your workers get the queues directly from a persisted database or an in memory technique ?
    – Ozum Safa
    Dec 27, 2016 at 12:13
  • The gathering might scale out if you establish predetermined rules for which match to accept in cases of multiple matches found, such as lowest IDs. This instead of relying on choosing the first to arrive by messaging. It would might take longer to inform the workers of elements that have been dismissed but at least the workers can be scaled further as needed.
    – Erik Eidt
    Dec 27, 2016 at 16:23
  • I was thinking along the lines of using simple messaging as in (micro) services and putting any necessary persistence more or less on the side.
    – Erik Eidt
    Dec 27, 2016 at 16:27
  • Even if you pick the lowest ID, if you have multiple gatherers, they have to be aware of all the IDs that has come to existence by then, which cannot be just solved by messaging system. Because, these IDs would need to be persisted somewhere(db or memory), and that again brings race condition. Also, messaging queue might be in insufficient, because any element in the queue can be matched with anyone else in teh queue. So it is not just a stream of work to be done. It could be used to receive the new elements to insert into the queue. But when actual matching is to be done...
    – Ozum Safa
    Dec 27, 2016 at 18:14
0

If you are implementing the queue in a database, you can achieve mutual update exclusion using the REPEATABLE READ transaction isolation level. The table will still allow inserts, but any rows that are read by your code will remain unmodifiable until your transaction has ended.

Thus if a worker wants to grab the next two available items, it can grab them with some simple SQL.

Say you have a table called Tasks with a primary key of ID and a column named WorkerID which is an identifier indicating which worker can work on the task. WorkerID is set to NULL to begin with. To grab the next two items you would issue the following command:

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

Naturally if your database is replicated, this sort of scheme would only would only work with transactional replication.

Improve this answer
0

Sounds like something I've done in the past. I'm assuming the queue is on one database instance.

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

You can have as many processes/threads as you like running this code. You are basically putting write locks on the records you are interested in and another transaction won't be able to grab that record. Also, because you're ordering your foreach using sequentialId, you can never get a deadlock condition.

Improve this answer
1
  • Locking the database is not as reliable when you don't have one instance. No production database has one instance, so this is kinda redundant.
    – Ozum Safa
    Feb 26, 2017 at 19:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.