Concurrency

GreyCat tasks are sequential but can spawn and wait jobs to create concurrent computation units.

A Job is a handle to a function and its arguments, creating a Job does nothing on its own, it is just a GreyCat object

Calling the await function will pause the current task while awaiting for jobs passed as arguments to complete.

fn long_computation(max: int): int {
  var count = 0;
  for (var i = 0; i < max; i++) {
    count++;
  }
  return count;
}

fn main() {
 // Define the task request object by providing the arguments and the function pointer
 var jobs = Array<Job> {
     Job {function: project::long_computation , arguments: [100_000] },
     Job {function: project::long_computation , arguments: [100_000] }
 };

// Blocks code execution awaiting the end or fails early if an error occurs
 await(jobs);

 for(_, job in jobs){
  // accessing the result
  var result = job.result();
 }
}

Jobs run in parallel on the worker pool, both in a server task and under greycat run. Only greycat test runs them one after the other.

Every job.result() stays readable until the task ends, also after later awaits. The results are kept in files of the task and each call reads one, so a large fan-out can be awaited chunk by chunk and its results read one at a time at the end. Keep the value in a variable rather than calling result() twice. A Job kept past its task (in a node, for instance) throws on result().

fn double(i: int): int {
  return i * 2;
}

fn main() {
  var all = Array<Job> {};
  for (var chunk = 0; chunk < 10; chunk++) {
    var jobs = Array<Job> {};
    for (var i = 0; i < 100; i++) {
      jobs.add(Job { function: project::double, arguments: [chunk * 100 + i] });
    }
    await(jobs);
    all.add_all(jobs);
  }
  var sum = 0;
  for (_, job in all) {
    sum = sum + job.result() as int;
  }
  println(sum);
}

Jobs inherits from the eventual modifications made from the parent task to any nodes.

Conversely, parent task will aggregate all modifications done by jobs right after the await method call.

GreyCat Task and Job model can be compared to a sequence of fork-and-join process.

Handling Errors in Jobs

await throws if at least one job has thrown, with the message and stack of the first failed job. Catch it, then read each job’s result: a job that threw returns its Error from result(). A job that returns an Error without throwing does not make await throw.

The following snippet demonstrates how you would handle a scenario where you don’t want the whole execution to fail if one Jobs fails.

fn main() {
  var jobs = Array<Job> {
    Job { function: foo, arguments: [10_s, false] },
    Job { function: foo, arguments: [1_s, true] }
  };
  try {
    await(jobs);
  } catch (err) {
    error(err);
  }
  for (i, job in jobs) {
    var res = job.result();
    if (res is Error) {
      println("Job ${i} failed");
    } else {
      println("Job ${i} finished");
    }
  }
}

fn foo(a: duration, fail: bool) {
  Runtime::sleep(a);
  if (fail) {
    throw "Failed";
  }
}

Parallel writes (Concurrent Writing)

You are able to write in parallel to the graph. When the writes of the jobs merge back into the parent task, a node or an object field changed by a single job takes that change. A nodeTime, nodeIndex, nodeList or nodeGeo written by several jobs merges key by key, and so does a Map, so jobs writing different keys never conflict. The same key, or the same field, written with different values by two jobs is a conflict. await then throws and keeps nothing it merged.

Two jobs appending to the same nodeList with add write the same key and collide. Give each job its own range of indexes with set, or return the new nodes and add them in the parent task as below.

The following code snippet shows an example of creating independent nodes in parallel and insert them into a global index at the end;

This model becomes very powerful when used to import big amounts of data from separate files, or executing heavy computation.

type Sensor{
  history:nodeTime<int>;
}
var sensor_list: nodeList<node<Sensor>>;

fn main() {
  var jobs = Array<Job> {};

  jobs.add(Job {function: project::import });
  jobs.add(Job {function: project::import });

  await(jobs);

  for (_, job in jobs) {
    var sensors = job.result();
    for (_, sensor: node<Sensor> in sensors) {
      sensor_list.add(sensor);
    }
  }
}

fn import(): Array<node<Sensor>> {
  var sensors = Array<node<Sensor>> {};
  for (var i = 0; i < 10; i++) {
    var sensor = node<Sensor> { Sensor { history: nodeTime<int> {} }};
    sensors.add(sensor);
  }
  return sensors;
}

Merge strategy

The same rules apply when a task commits over another task that committed first. By default a conflict fails the commit with incompatible fusion, and the task ends in error with its writes rolled back.

A task can choose how its own commit resolves conflicts with Task::set_merge_strategy:

• MergeStrategy::fusion_by_key_fail_on_conflict, the default, merges key by key and fails on a conflict.
• MergeStrategy::fusion_by_key_last_win merges key by key, and a conflicting key or field takes the value this task wrote.
• MergeStrategy::no_fusion_fail_on_conflict fails the commit, or the await that merges the jobs, when a nodeTime, nodeIndex, nodeList or nodeGeo was changed on both sides. Plain fields and objects still merge field by field.

fn refresh_cache() {
  Task::set_merge_strategy(MergeStrategy::fusion_by_key_last_win);
  // on a conflict, the value written by this task wins
}

The strategy belongs to the task and covers every write it commits, including those of the functions it calls. It is not inherited by the tasks it spawns. Keep fusion_by_key_last_win for tasks whose writes are all cache-like, since it silently overrides concurrent changes.

Limitations

Take the following snippet as an example, it will raise the following error message wrong state before await, variable contains an object stored in a node.

type Foo {
  status: String;
}

fn task(foo: node<Foo>) {
  var resolved_foo = foo.resolve();

  // awaiting jobs

  resolved_foo.status = "Done";
}

When execution reaches an await point, the current function scope is serialized.

This means that any object resolved from a node before an await may become outdated after resumption, as the node’s content might have changed in the meantime.

Accessing such outdated values is unsafe and will result in an error.

Scope serialization occurs because the current execution context is discarded and later reconstructed from the serialized data upon resuming execution after the await.

Leverage the arrow operator if you need to modify an object inside a node, or set the variable to null before the await, resolved_foo = null, both methods will work.

A writer opened before an await does not keep its file across it. Written again after the await, it truncates the file. Write the rest with a new writer created with append: true.