1

I have a data synchronisation concurrent algorithm. It does the following: get data and files from server, send data and files to server, save them to database / filesystem. Imagine the system like this:

  1. You have 1000 functions. Each one does some atomic operation. For instance, fetch latest objects of type X and insert them into DB; upload this file of type Y and so on. Each function is independent and can act on its own, it does not communicate with or affect other functions. On the other hand, none of them is a pure function, because they all use theese common resources (fetching data from the server, puting data on DB, saving files on filesystem)
  2. You have a single entry point for the sychronization mechanism. The outside of the sync system can start the sync, say, by doing a Sync.start() call. Also, the sync has a single exit point. The sync can finish with either success, either failure (if any of those functions from (1) fail, the whole sync will fail). The ouside of the sync system can subscribe to onSyncSuccess / onSyncError events.
  3. You have this black box in the middle of the system. This could be, for instance, a single threaded algorithm calling those 1000 functions from (1). But I made it concurrent.

Now consider this. This concurrent algorithm right now is rigid because the way in which the functions are called is hardcoded. If I want to take a bunch of functions from (1) that right now are executing sequentially, and if I want to make them execute parallel, it would be impossible without refactoring the whole class hierarchy.

I was thinking about the concept of direct acyclic graphs, and I made my own domain-specific language in Kotlin to define such task graphs. Now I could write the whole orchestration declaratively like this:

notifySyncWasStarted()
runSequentialy {
    task { doTask1() }
    runInParallel {
        task { doTask2() }
        task { doTask3() }
    }
    task { doTask4() }
}
notifySyncWasStopped()

So first task1 gets executed, then task2 and 3 in the same time, then task4. By keeping this graph in a single file, I could easily modify the way tasks are executed. For instance, I could easily swap tasks:

notifySyncWasStarted()
runSequentialy {
    runInParallel {
        task { doTask4() }
        task { doTask2() }
    }
    task { doTask3() }
    task { doTask1() }
}
notifySyncWasStopped()

Here, (task 4 and 2) gets executed, then 3, then 1. This works by using the fork-join paradigm, I create threads then join them into the parent thread.

In contrast, right now, the algorithm is spread around multiple classes, each of them was designed to run the tasks in a specific manner. Changing how tasks are ran would mean to refactor the classes and how they communicate to each other.

The question is: What is the best way to decouple and define the orchestration (coordination) of concurrent tasks? So that this orchestration could be easily changed in the future? Is my solution optimal or the way to go (direct acyclic graphs, fork-join, plus a domain specific language)? Or maybe there are some other design patterns that do the same thing?

New contributor
user1658358 is a new contributor to this site. Take care in asking for clarification, commenting, and answering. Check out our Code of Conduct.
  • I fail to see what direct acyclic graphs are doing for you here. I do like how your DSL doesn't dictate your concurrency implementation and only states the algorithms needs. I'm curious if notifySyncWasStarted() is supposed to stop the rest of the servers from updating while this runs. – candied_orange 2 days ago
  • well, I don't store graph data per se, but in theory, the algorithm, as described with the DSL, can be represented as a graph. I know this is the exact terminology gradle is using, "DAG of tasks". No, notify() don't block anything, those are just signaling some events to subscribers via a non-blocking event bus. – user1658358 2 days ago
  • 1
    Leaves me curious how you deal with the shared mutable state of the systems you’re talking to. – candied_orange 2 days ago
  • First you wrote "Each function is independent and can act on its own, it does not communicate with or affect other functions." which means to me, order of execution is completely irrelevant. So why do you need something to describe constraints on the order of execution at all? – Doc Brown 2 days ago
  • @Doc Brown yeah, I missed this detail. some things need to go in a specific order. most of them not. I mean, those functions are not communicating directly, but they do all have access to a shared database. candied_orange well, there is no shared mutable state, except the local database and the remote database (via rest apis) – user1658358 2 days ago
4

This sounds like the role of workflow software such as FireWorks.

(The term "workflow management" can also refer to managing the handoffs of operations to perform between people.)

There are hundreds of workflow systems available. That may be a sign that the domain is insufficiently understood and a sign that the applications, runtime environments, and scales vary widely.

Often the workflow does indeed get described as a Directed Acyclic Graph. In FireWorks you can write it as a .yaml file or Python code, or implement a workflow builder in Python code that translates application parameters to a specific DAG.

In my limited experience, the DAG is at a higher level than your example: Rather than specifying what to run in parallel vs. series, the DAG specifies the dependencies between tasks. Then the workflow software will mark each task "ready" to run as soon as all of its input (or "upstream") dependencies have been successfully met. Each "ready" task will then run once the computing resources are available -- there may be additional factors like task priorities.

  • I've found that FireWorks is pretty straightforward, well documented, actively maintained, and flexible. Tasks can run conditionally and can even modify the workflow DAG. It only requires one server -- a MongoDB instance.
  • In comparison, Apache Airflow has many moving parts so it helps that it's available preconfigured as Google Cloud Composer.

Q. How to decouple and define the workflow orchestration so that it can be easily changed in the future?

A. I'd build on the design or the implementation of one of these existing workflow systems and its DAG-building language. These projects have spent years evolving the workflow approach to aid real-world applications.

| improve this answer | |
  • nice. thanks. I will dig this concept. This algorithm is actually used inside a native android app, so I need to find a lightweight, embeddable java / kotlin workflow library, somthing like github.com/j-easy/easy-flows – user1658358 yesterday

Your Answer

user1658358 is a new contributor. Be nice, and check out our Code of Conduct.

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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